



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      _UpperCAmelCase      (							A__  ):
     UpperCamelCase__		     =					42







class      _UpperCAmelCase      (							A__      ,   A__  ):




     @register_to_config
     def __init__(					self			,    a__ = 3_2			,    a__ = 6_4			,    a__ = 2_0			,    a__ = 7_6_8			,    a__=7_7			,    a__=4			,    a__ = 0.0			,    a__ = "silu"			,    a__ = None			,    a__ = None			,    a__ = "linear"			,    a__ = "prd"			,    a__ = None			,    a__ = None			,    a__ = None			,    ):
         super().__init__()
         A__					=       num_attention_heads
         A__					=       attention_head_dim
         A__					=       num_attention_heads * attention_head_dim
         A__					=       additional_embeddings

         A__					=       time_embed_dim or inner_dim
         A__					=       embedding_proj_dim or embedding_dim
         A__					=       clip_embed_dim or embedding_dim

         A__					=       Timesteps(a__			,    a__			,    0)
         A__					=       TimestepEmbedding(a__			,    a__			,    out_dim=a__			,    act_fn=a__)

         A__					=       nn.Linear(a__			,    a__)

         if embedding_proj_norm_type is None:
             A__					=       None
         elif embedding_proj_norm_type == "layer":
             A__					=       nn.LayerNorm(a__)
         else:
             raise ValueError(F"unsupported embedding_proj_norm_type: {embedding_proj_norm_type}")

         A__					=       nn.Linear(a__			,    a__)

         if encoder_hid_proj_type is None:
             A__					=       None
         elif encoder_hid_proj_type == "linear":
             A__					=       nn.Linear(a__			,    a__)
         else:
             raise ValueError(F"unsupported encoder_hid_proj_type: {encoder_hid_proj_type}")

         A__					=       nn.Parameter(torch.zeros(1			,    num_embeddings + additional_embeddings			,    a__))

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

         A__					=       nn.ModuleList(
             [
                 BasicTransformerBlock(
                     a__			,    a__			,    a__			,    dropout=a__			,    activation_fn='''gelu'''			,    attention_bias=a__			,    )
                 for d in range(a__)
             ])

         if norm_in_type == "layer":
             A__					=       nn.LayerNorm(a__)
         elif norm_in_type is None:
             A__					=       None
         else:
             raise ValueError(F"Unsupported norm_in_type: {norm_in_type}.")

         A__					=       nn.LayerNorm(a__)

         A__					=       nn.Linear(a__			,    a__)

         A__					=       torch.full(
             [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]			,    -1_0_0_0_0.0)
         causal_attention_mask.triu_(1)
         A__					=       causal_attention_mask[None, ...]
         self.register_buffer('''causal_attention_mask'''			,    a__			,    persistent=a__)

         A__					=       nn.Parameter(torch.zeros(1			,    a__))
         A__					=       nn.Parameter(torch.zeros(1			,    a__))




     @property
     # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
     def 		snake_case_		(					self):
         A__					=       {}

         def fn_recursive_add_processors(a__			,    a__			,    a__):
             if hasattr(a__			,    '''set_processor'''):
                 A__					=       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 		snake_case_		(					self			,    a__):
         A__					=       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__			,    a__			,    a__):
             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 		snake_case_		(					self):
         self.set_attn_processor(AttnProcessor())




     def 		snake_case_		(					self			,    a__			,    a__			,    a__			,    a__ = None			,    a__ = None			,    a__ = True			,    ):
         A__					=       hidden_states.shape[0]

         A__					=       timestep
         if not torch.is_tensor(a__):
             A__					=       torch.tensor([timesteps]			,    dtype=torch.long			,    device=hidden_states.device)
         elif torch.is_tensor(a__) and len(timesteps.shape) == 0:
             A__					=       timesteps[None].to(hidden_states.device)

         # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
         A__					=       timesteps * torch.ones(a__			,    dtype=timesteps.dtype			,    device=timesteps.device)

         A__					=       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.
         A__					=       timesteps_projected.to(dtype=self.dtype)
         A__					=       self.time_embedding(a__)

         if self.embedding_proj_norm is not None:
             A__					=       self.embedding_proj_norm(a__)

         A__					=       self.embedding_proj(a__)
         if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
             A__					=       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''')

         A__					=       self.proj_in(a__)

         A__					=       self.positional_embedding.to(hidden_states.dtype)

         A__					=       []
         A__					=       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:
             A__					=       proj_embeddings[:, None, :]

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

         A__					=       additional_embeds + [
             proj_embeddings,
             time_embeddings[:, None, :],
             hidden_states,
         ]

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

         A__					=       torch.cat(
             a__			,    dim=1			,    )

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

         A__					=       hidden_states + positional_embeddings

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

         if self.norm_in is not None:
             A__					=       self.norm_in(a__)

         for block in self.transformer_blocks:
             A__					=       block(a__			,    attention_mask=a__)

         A__					=       self.norm_out(a__)

         if self.prd_embedding is not None:
             A__					=       hidden_states[:, -1]
         else:
             A__					=       hidden_states[:, additional_embeddings_len:]

         A__					=       self.proj_to_clip_embeddings(a__)

         if not return_dict:
             return (predicted_image_embedding,)

         return PriorTransformerOutput(predicted_image_embedding=a__)




     def 		snake_case_		(					self			,    a__):
         A__					=       (prior_latents * self.clip_std) + self.clip_mean
         return prior_latents





from typing import TYPE_CHECKING

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


_lowercase    	=		{
    "configuration_mctct": ["MCTCT_PRETRAINED_CONFIG_ARCHIVE_MAP", "MCTCTConfig"],
    "feature_extraction_mctct": ["MCTCTFeatureExtractor"],
    "processing_mctct": ["MCTCTProcessor"],
}


try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   _lowercase    	=		[
       "MCTCT_PRETRAINED_MODEL_ARCHIVE_LIST",
       "MCTCTForCTC",
       "MCTCTModel",
       "MCTCTPreTrainedModel",
   ]


if TYPE_CHECKING:
   from .configuration_mctct import MCTCT_PRETRAINED_CONFIG_ARCHIVE_MAP, MCTCTConfig
   from .feature_extraction_mctct import MCTCTFeatureExtractor
   from .processing_mctct import MCTCTProcessor

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_mctct import MCTCT_PRETRAINED_MODEL_ARCHIVE_LIST, MCTCTForCTC, MCTCTModel, MCTCTPreTrainedModel

else:
   import sys

   _lowercase    	=		_LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)
