



'''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_    (				SCREAMING_SNAKE_CASE_     ):
  """simple docstring"""




  a_												=42






class 					lowerCamelCase_    (				SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_     ):
  """simple docstring"""




  @register_to_config
  def __init__( self      :	Optional[int]			,	_a      :	int = 32			,	_a      :	int = 64			,	_a      :	int = 20			,	_a      :	int = 768			,	_a      :	int=77			,	_a      :	Tuple=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			,	)					->					Optional[int]:
    super().__init__()
    __lowerCamelCase       :       Union[str, Any]           = num_attention_heads
    __lowerCamelCase       :       Any           = attention_head_dim
    __lowerCamelCase       :       List[str]           = num_attention_heads * attention_head_dim
    __lowerCamelCase       :       int           = additional_embeddings

    __lowerCamelCase       :       Optional[Any]           = time_embed_dim or inner_dim
    __lowerCamelCase       :       Dict           = embedding_proj_dim or embedding_dim
    __lowerCamelCase       :       Tuple           = clip_embed_dim or embedding_dim

    __lowerCamelCase       :       Union[str, Any]           = Timesteps(_a			,	_a			,	0   )
    __lowerCamelCase       :       Tuple           = TimestepEmbedding(_a			,	_a			,	out_dim=_a			,	act_fn=_a   )

    __lowerCamelCase       :       int           = nn.Linear(_a			,	_a   )

    if embedding_proj_norm_type is None:
      __lowerCamelCase       :       Optional[Any]           = None
    elif embedding_proj_norm_type == "layer":
      __lowerCamelCase       :       Dict           = nn.LayerNorm(_a   )
    else:
      raise ValueError(f'unsupported embedding_proj_norm_type: {embedding_proj_norm_type}'   )

    __lowerCamelCase       :       List[str]           = nn.Linear(_a			,	_a   )

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

    __lowerCamelCase       :       str           = nn.Parameter(torch.zeros(1			,	num_embeddings + additional_embeddings			,	_a   )   )

    if added_emb_type == "prd":
      __lowerCamelCase       :       Optional[Any]           = nn.Parameter(torch.zeros(1			,	1			,	_a   )   )
    elif added_emb_type is None:
      __lowerCamelCase       :       List[Any]           = None
    else:
      raise ValueError(
          f'`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `\'prd\'` or `None`.'   )

    __lowerCamelCase       :       int           = nn.ModuleList(
        [
            BasicTransformerBlock(
                _a			,	_a			,	_a			,	dropout=_a			,	activation_fn='gelu'			,	attention_bias=_a			,	)
            for d in range(_a   )
        ]   )

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

    __lowerCamelCase       :       List[Any]           = nn.LayerNorm(_a   )

    __lowerCamelCase       :       Tuple           = nn.Linear(_a			,	_a   )

    __lowerCamelCase       :       int           = torch.full(
        [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]			,	-10000.0   )
    causal_attention_mask.triu_(1   )
    __lowerCamelCase       :       List[str]           = causal_attention_mask[None, ...]
    self.register_buffer('causal_attention_mask'			,	_a			,	persistent=_a   )

    __lowerCamelCase       :       Optional[Any]           = nn.Parameter(torch.zeros(1			,	_a   )   )
    __lowerCamelCase       :       List[str]           = nn.Parameter(torch.zeros(1			,	_a   )   )




  @property
  # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
  def 						_lowercase     ( self      :	Optional[int]   )					->					Dict[str, AttentionProcessor]:
    __lowerCamelCase       :       Union[str, Any]           = {}

    def fn_recursive_add_processors(_a      :	str			,	_a      :	torch.nn.Module			,	_a      :	Dict[str, AttentionProcessor]   ):
      if hasattr(_a			,	'set_processor'   ):
        __lowerCamelCase       :       Union[str, Any]           = 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      :	int			,	_a      :	Union[AttentionProcessor, Dict[str, AttentionProcessor]]   )					->					Union[str, Any]:
    __lowerCamelCase       :       List[str]           = 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      :	int   ):
      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      :	Tuple   )					->					List[Any]:
    self.set_attn_processor(AttnProcessor()   )




  def 						_lowercase     ( self      :	int			,	_a      :	Any			,	_a      :	Union[torch.Tensor, float, int]			,	_a      :	torch.FloatTensor			,	_a      :	Optional[torch.FloatTensor] = None			,	_a      :	Optional[torch.BoolTensor] = None			,	_a      :	bool = True			,	)					->					int:
    __lowerCamelCase       :       List[Any]           = hidden_states.shape[0]

    __lowerCamelCase       :       List[Any]           = timestep
    if not torch.is_tensor(_a   ):
      __lowerCamelCase       :       Dict           = torch.tensor([timesteps]			,	dtype=torch.long			,	device=hidden_states.device   )
    elif torch.is_tensor(_a   ) 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       :       Optional[Any]           = timesteps * torch.ones(_a			,	dtype=timesteps.dtype			,	device=timesteps.device   )

    __lowerCamelCase       :       List[str]           = 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.
    __lowerCamelCase       :       List[str]           = timesteps_projected.to(dtype=self.dtype   )
    __lowerCamelCase       :       Dict           = self.time_embedding(_a   )

    if self.embedding_proj_norm is not None:
      __lowerCamelCase       :       List[str]           = self.embedding_proj_norm(_a   )

    __lowerCamelCase       :       Optional[int]           = self.embedding_proj(_a   )
    if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
      __lowerCamelCase       :       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'   )

    __lowerCamelCase       :       Optional[Any]           = self.proj_in(_a   )

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

    __lowerCamelCase       :       Any           = []
    __lowerCamelCase       :       str           = 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:
      __lowerCamelCase       :       int           = proj_embeddings[:, None, :]

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

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

    if self.prd_embedding is not None:
      __lowerCamelCase       :       Dict           = self.prd_embedding.to(hidden_states.dtype   ).expand(_a			,	-1			,	-1   )
      additional_embeds.append(_a   )

    __lowerCamelCase       :       Optional[Any]           = torch.cat(
        _a			,	dim=1			,	)

    # Allow positional_embedding to not include the `addtional_embeddings` and instead pad it with zeros for these additional tokens
    __lowerCamelCase       :       Union[str, Any]           = additional_embeddings_len + proj_embeddings.shape[1] + 1
    if positional_embeddings.shape[1] < hidden_states.shape[1]:
      __lowerCamelCase       :       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			,	)

    __lowerCamelCase       :       Tuple           = hidden_states + positional_embeddings

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

    if self.norm_in is not None:
      __lowerCamelCase       :       List[str]           = self.norm_in(_a   )

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

    __lowerCamelCase       :       Union[str, Any]           = self.norm_out(_a   )

    if self.prd_embedding is not None:
      __lowerCamelCase       :       Union[str, Any]           = hidden_states[:, -1]
    else:
      __lowerCamelCase       :       Optional[Any]           = hidden_states[:, additional_embeddings_len:]

    __lowerCamelCase       :       int           = self.proj_to_clip_embeddings(_a   )

    if not return_dict:
      return (predicted_image_embedding,)

    return PriorTransformerOutput(predicted_image_embedding=_a   )




  def 						_lowercase     ( self      :	List[Any]			,	_a      :	str   )					->					Dict:
    __lowerCamelCase       :       Optional[int]           = (prior_latents * self.clip_std) + self.clip_mean
    return prior_latents











'''simple docstring'''





from typing import TYPE_CHECKING

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


_UpperCamelCase											=						{'configuration_speech_encoder_decoder': ['SpeechEncoderDecoderConfig']}

try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   _UpperCamelCase											=						['SpeechEncoderDecoderModel']

try:
   if not is_flax_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   _UpperCamelCase											=						['FlaxSpeechEncoderDecoderModel']

if TYPE_CHECKING:
   from .configuration_speech_encoder_decoder import SpeechEncoderDecoderConfig

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_speech_encoder_decoder import SpeechEncoderDecoderModel

   try:
      if not is_flax_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_flax_speech_encoder_decoder import FlaxSpeechEncoderDecoderModel

else:
   import sys

   _UpperCamelCase											=						_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)






