




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_       (					lowerCamelCase_							):
				__a     :       torch.FloatTensor







class 				lowerCAmelCase_       (					lowerCamelCase_						,     lowerCamelCase_							):






				@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__()
							SCREAMING_SNAKE_CASE_   : Any					  =      num_attention_heads
							SCREAMING_SNAKE_CASE_   : str					  =      attention_head_dim
							SCREAMING_SNAKE_CASE_   : Dict					  =      num_attention_heads * attention_head_dim
							SCREAMING_SNAKE_CASE_   : str					  =      additional_embeddings

							SCREAMING_SNAKE_CASE_   : str					  =      time_embed_dim or inner_dim
							SCREAMING_SNAKE_CASE_   : int					  =      embedding_proj_dim or embedding_dim
							SCREAMING_SNAKE_CASE_   : str					  =      clip_embed_dim or embedding_dim

							SCREAMING_SNAKE_CASE_   : Tuple					  =      Timesteps(snake_case__      ,snake_case__      ,0 )
							SCREAMING_SNAKE_CASE_   : Optional[int]					  =      TimestepEmbedding(snake_case__      ,snake_case__      ,out_dim=snake_case__      ,act_fn=snake_case__ )

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

							if embedding_proj_norm_type is None:
										SCREAMING_SNAKE_CASE_   : int					  =      None
							elif embedding_proj_norm_type == "layer":
										SCREAMING_SNAKE_CASE_   : List[Any]					  =      nn.LayerNorm(snake_case__ )
							else:
										raise ValueError(F'unsupported embedding_proj_norm_type: {embedding_proj_norm_type}' )

							SCREAMING_SNAKE_CASE_   : Optional[int]					  =      nn.Linear(snake_case__      ,snake_case__ )

							if encoder_hid_proj_type is None:
										SCREAMING_SNAKE_CASE_   : Tuple					  =      None
							elif encoder_hid_proj_type == "linear":
										SCREAMING_SNAKE_CASE_   : str					  =      nn.Linear(snake_case__      ,snake_case__ )
							else:
										raise ValueError(F'unsupported encoder_hid_proj_type: {encoder_hid_proj_type}' )

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

							if added_emb_type == "prd":
										SCREAMING_SNAKE_CASE_   : str					  =      nn.Parameter(torch.zeros(1      ,1      ,snake_case__ ) )
							elif added_emb_type is None:
										SCREAMING_SNAKE_CASE_   : Union[str, Any]					  =      None
							else:
										raise ValueError(
										    F'`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `\'prd\'` or `None`.' )

							SCREAMING_SNAKE_CASE_   : Tuple					  =      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":
										SCREAMING_SNAKE_CASE_   : Dict					  =      nn.LayerNorm(snake_case__ )
							elif norm_in_type is None:
										SCREAMING_SNAKE_CASE_   : str					  =      None
							else:
										raise ValueError(F'Unsupported norm_in_type: {norm_in_type}.' )

							SCREAMING_SNAKE_CASE_   : int					  =      nn.LayerNorm(snake_case__ )

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

							SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      torch.full(
							    [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]      ,-10000.0 )
							causal_attention_mask.triu_(1 )
							SCREAMING_SNAKE_CASE_   : Tuple					  =      causal_attention_mask[None, ...]
							self.register_buffer('causal_attention_mask'      ,snake_case__      ,persistent=snake_case__ )

							SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      nn.Parameter(torch.zeros(1      ,snake_case__ ) )
							SCREAMING_SNAKE_CASE_   : Tuple					  =      nn.Parameter(torch.zeros(1      ,snake_case__ ) )






				@property
				# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
				def 		snake_case	(			self ):
							SCREAMING_SNAKE_CASE_   : Dict					  =      {}

							def fn_recursive_add_processors(snake_case__      ,snake_case__      ,snake_case__ ):
										if hasattr(snake_case__      ,'set_processor' ):
													SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      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 		snake_case	(			self      ,snake_case__ ):
							SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      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 		snake_case	(			self ):
							self.set_attn_processor(AttnProcessor() )






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

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

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

							SCREAMING_SNAKE_CASE_   : 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.
							SCREAMING_SNAKE_CASE_   : int					  =      timesteps_projected.to(dtype=self.dtype )
							SCREAMING_SNAKE_CASE_   : Any					  =      self.time_embedding(snake_case__ )

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

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

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

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

							SCREAMING_SNAKE_CASE_   : Any					  =      []
							SCREAMING_SNAKE_CASE_   : List[Any]					  =      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:
										SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      proj_embeddings[:, None, :]

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

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

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

							SCREAMING_SNAKE_CASE_   : str					  =      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
							SCREAMING_SNAKE_CASE_   : List[Any]					  =      additional_embeddings_len + proj_embeddings.shape[1] + 1
							if positional_embeddings.shape[1] < hidden_states.shape[1]:
										SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      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      ,)

							SCREAMING_SNAKE_CASE_   : int					  =      hidden_states + positional_embeddings

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

							if self.norm_in is not None:
										SCREAMING_SNAKE_CASE_   : List[str]					  =      self.norm_in(snake_case__ )

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

							SCREAMING_SNAKE_CASE_   : List[str]					  =      self.norm_out(snake_case__ )

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

							SCREAMING_SNAKE_CASE_   : str					  =      self.proj_to_clip_embeddings(snake_case__ )

							if not return_dict:
										return (predicted_image_embedding,)

							return PriorTransformerOutput(predicted_image_embedding=snake_case__ )






				def 		snake_case	(			self      ,snake_case__ ):
							SCREAMING_SNAKE_CASE_   : Optional[Any]					  =      (prior_latents * self.clip_std) + self.clip_mean
							return prior_latents




from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_torch_available,
)


lowerCamelCase__  =					{
    '''configuration_wav2vec2''': ['''WAV_2_VEC_2_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''Wav2Vec2Config'''],
    '''feature_extraction_wav2vec2''': ['''Wav2Vec2FeatureExtractor'''],
    '''processing_wav2vec2''': ['''Wav2Vec2Processor'''],
    '''tokenization_wav2vec2''': ['''Wav2Vec2CTCTokenizer''', '''Wav2Vec2Tokenizer'''],
}


try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   lowerCamelCase__  =					[
       '''WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST''',
       '''Wav2Vec2ForAudioFrameClassification''',
       '''Wav2Vec2ForCTC''',
       '''Wav2Vec2ForMaskedLM''',
       '''Wav2Vec2ForPreTraining''',
       '''Wav2Vec2ForSequenceClassification''',
       '''Wav2Vec2ForXVector''',
       '''Wav2Vec2Model''',
       '''Wav2Vec2PreTrainedModel''',
   ]

try:
   if not is_tf_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   lowerCamelCase__  =					[
       '''TF_WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST''',
       '''TFWav2Vec2ForCTC''',
       '''TFWav2Vec2Model''',
       '''TFWav2Vec2PreTrainedModel''',
       '''TFWav2Vec2ForSequenceClassification''',
   ]

try:
   if not is_flax_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   lowerCamelCase__  =					[
       '''FlaxWav2Vec2ForCTC''',
       '''FlaxWav2Vec2ForPreTraining''',
       '''FlaxWav2Vec2Model''',
       '''FlaxWav2Vec2PreTrainedModel''',
   ]


if TYPE_CHECKING:
   from .configuration_wavaveca import WAV_2_VEC_2_PRETRAINED_CONFIG_ARCHIVE_MAP, WavaVecaConfig
   from .feature_extraction_wavaveca import WavaVecaFeatureExtractor
   from .processing_wavaveca import WavaVecaProcessor
   from .tokenization_wavaveca import WavaVecaCTCTokenizer, WavaVecaTokenizer

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_wavaveca import (
          WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST,
          WavaVecaForAudioFrameClassification,
          WavaVecaForCTC,
          WavaVecaForMaskedLM,
          WavaVecaForPreTraining,
          WavaVecaForSequenceClassification,
          WavaVecaForXVector,
          WavaVecaModel,
          WavaVecaPreTrainedModel,
      )

   try:
      if not is_tf_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_tf_wavaveca import (
          TF_WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST,
          TFWavaVecaForCTC,
          TFWavaVecaForSequenceClassification,
          TFWavaVecaModel,
          TFWavaVecaPreTrainedModel,
      )

   try:
      if not is_flax_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_tf_wavaveca import (
          FlaxWavaVecaForCTC,
          FlaxWavaVecaForPreTraining,
          FlaxWavaVecaModel,
          FlaxWavaVecaPreTrainedModel,
      )


else:
   import sys

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




