

from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..utils import maybe_allow_in_graph
from .activations import get_activation
from .attention_processor import Attention
from .embeddings import CombinedTimestepLabelEmbeddings






@maybe_allow_in_graph
class      lowerCamelCase_      (	nn.Module	):




	def __init__(     self						:     Tuple					,      __A						:     int					,      __A						:     int					,      __A						:     int					,      __A						:     str=0.0					,      __A						:     Optional[int] = None					,      __A						:     str = "geglu"					,      __A						:     Optional[int] = None					,      __A						:     bool = False					,      __A						:     bool = False					,      __A						:     bool = False					,      __A						:     bool = False					,      __A						:     bool = True					,      __A						:     str = "layer_norm"					,      __A						:     bool = False					,      ):
		super().__init__()
		__A :			Any		    =			only_cross_attention

		__A :			Dict		    =			(num_embeds_ada_norm is not None) and norm_type == """ada_norm_zero"""
		__A :			Optional[Any]		    =			(num_embeds_ada_norm is not None) and norm_type == """ada_norm"""

		if norm_type in ("ada_norm", "ada_norm_zero") and num_embeds_ada_norm is None:
			raise ValueError(
			    F"""`norm_type` is set to {norm_type}, but `num_embeds_ada_norm` is not defined. Please make sure to"""
			    F""" define `num_embeds_ada_norm` if setting `norm_type` to {norm_type}."""							)

		# Define 3 blocks. Each block has its own normalization layer.
		# 1. Self-Attn
		if self.use_ada_layer_norm:
			__A :			str		    =			AdaLayerNorm(__A					,      __A							)
		elif self.use_ada_layer_norm_zero:
			__A :			Dict		    =			AdaLayerNormZero(__A					,      __A							)
		else:
			__A :			Optional[int]		    =			nn.LayerNorm(__A					,      elementwise_affine=__A							)
		__A :			List[str]		    =			Attention(
		    query_dim=__A					,      heads=__A					,      dim_head=__A					,      dropout=__A					,      bias=__A					,      cross_attention_dim=cross_attention_dim if only_cross_attention else None					,      upcast_attention=__A					,      )

		# 2. Cross-Attn
		if cross_attention_dim is not None or double_self_attention:
			# We currently only use AdaLayerNormZero for self attention where there will only be one attention block.
			# I.e. the number of returned modulation chunks from AdaLayerZero would not make sense if returned during
			# the second cross attention block.
			__A :			Optional[Any]		    =			(
			    AdaLayerNorm(__A					,      __A							)
			    if self.use_ada_layer_norm
			    else nn.LayerNorm(__A					,      elementwise_affine=__A							)
			)
			__A :			Any		    =			Attention(
			    query_dim=__A					,      cross_attention_dim=cross_attention_dim if not double_self_attention else None					,      heads=__A					,      dim_head=__A					,      dropout=__A					,      bias=__A					,      upcast_attention=__A					,      )  # is self-attn if encoder_hidden_states is none
		else:
			__A :			str		    =			None
			__A :			int		    =			None

		# 3. Feed-forward
		__A :			Optional[int]		    =			nn.LayerNorm(__A					,      elementwise_affine=__A							)
		__A :			Dict		    =			FeedForward(__A					,      dropout=__A					,      activation_fn=__A					,      final_dropout=__A							)

		# let chunk size default to None
		__A :			List[Any]		    =			None
		__A :			Union[str, Any]		    =			0




	def        lowerCAmelCase_       (     self						:     List[str]					,      __A						:     Optional[int]					,      __A						:     int							):
		# Sets chunk feed-forward
		__A :			str		    =			chunk_size
		__A :			List[str]		    =			dim




	def        lowerCAmelCase_       (     self						:     Optional[int]					,      __A						:     torch.FloatTensor					,      __A						:     Optional[torch.FloatTensor] = None					,      __A						:     Optional[torch.FloatTensor] = None					,      __A						:     Optional[torch.FloatTensor] = None					,      __A						:     Optional[torch.LongTensor] = None					,      __A						:     Dict[str, Any] = None					,      __A						:     Optional[torch.LongTensor] = None					,      ):
		# Notice that normalization is always applied before the real computation in the following blocks.
		# 1. Self-Attention
		if self.use_ada_layer_norm:
			__A :			Tuple		    =			self.norma(__A					,      __A							)
		elif self.use_ada_layer_norm_zero:
			__A    ,				__A    ,				__A    ,				__A    ,				__A :			Union[str, Any]		    =			self.norma(
			    __A					,      __A					,      __A					,      hidden_dtype=hidden_states.dtype							)
		else:
			__A :			List[str]		    =			self.norma(__A							)

		__A :			Tuple		    =			cross_attention_kwargs if cross_attention_kwargs is not None else {}

		__A :			Optional[int]		    =			self.attna(
		    __A					,      encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None					,      attention_mask=__A					,      **__A					,      )
		if self.use_ada_layer_norm_zero:
			__A :			List[Any]		    =			gate_msa.unsqueeze(1							) * attn_output
		__A :			Optional[int]		    =			attn_output + hidden_states

		# 2. Cross-Attention
		if self.attna is not None:
			__A :			Union[str, Any]		    =			(
			    self.norma(__A					,      __A							) if self.use_ada_layer_norm else self.norma(__A							)
			)

			__A :			Tuple		    =			self.attna(
			    __A					,      encoder_hidden_states=__A					,      attention_mask=__A					,      **__A					,      )
			__A :			Union[str, Any]		    =			attn_output + hidden_states

		# 3. Feed-forward
		__A :			Tuple		    =			self.norma(__A							)

		if self.use_ada_layer_norm_zero:
			__A :			Tuple		    =			norm_hidden_states * (1 + scale_mlp[:, None]) + shift_mlp[:, None]

		if self._chunk_size is not None:
			# "feed_forward_chunk_size" can be used to save memory
			if norm_hidden_states.shape[self._chunk_dim] % self._chunk_size != 0:
				raise ValueError(
				    F"""`hidden_states` dimension to be chunked: {norm_hidden_states.shape[self._chunk_dim]} has to be divisible by chunk size: {self._chunk_size}. Make sure to set an appropriate `chunk_size` when calling `unet.enable_forward_chunking`."""							)

			__A :			str		    =			norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
			__A :			List[str]		    =			torch.cat(
			    [self.ff(__A							) for hid_slice in norm_hidden_states.chunk(__A					,      dim=self._chunk_dim							)]					,      dim=self._chunk_dim					,      )
		else:
			__A :			List[Any]		    =			self.ff(__A							)

		if self.use_ada_layer_norm_zero:
			__A :			List[Any]		    =			gate_mlp.unsqueeze(1							) * ff_output

		__A :			Any		    =			ff_output + hidden_states

		return hidden_states






class      lowerCamelCase_      (	nn.Module	):




	def __init__(     self						:     Union[str, Any]					,      __A						:     int					,      __A						:     Optional[int] = None					,      __A						:     int = 4					,      __A						:     float = 0.0					,      __A						:     str = "geglu"					,      __A						:     bool = False					,      ):
		super().__init__()
		__A :			Any		    =			int(dim * mult							)
		__A :			Optional[int]		    =			dim_out if dim_out is not None else dim

		if activation_fn == "gelu":
			__A :			Tuple		    =			GELU(__A					,      __A							)
		if activation_fn == "gelu-approximate":
			__A :			int		    =			GELU(__A					,      __A					,      approximate="""tanh"""							)
		elif activation_fn == "geglu":
			__A :			List[str]		    =			GEGLU(__A					,      __A							)
		elif activation_fn == "geglu-approximate":
			__A :			Any		    =			ApproximateGELU(__A					,      __A							)

		__A :			Optional[int]		    =			nn.ModuleList([]							)
		# project in
		self.net.append(__A							)
		# project dropout
		self.net.append(nn.Dropout(__A							)							)
		# project out
		self.net.append(nn.Linear(__A					,      __A							)							)
		# FF as used in Vision Transformer, MLP-Mixer, etc. have a final dropout
		if final_dropout:
			self.net.append(nn.Dropout(__A							)							)




	def        lowerCAmelCase_       (     self						:     Any					,      __A						:     Union[str, Any]							):
		for module in self.net:
			__A :			List[Any]		    =			module(__A							)
		return hidden_states






class      lowerCamelCase_      (	nn.Module	):




	def __init__(     self						:     Optional[Any]					,      __A						:     int					,      __A						:     int					,      __A						:     str = "none"							):
		super().__init__()
		__A :			Dict		    =			nn.Linear(__A					,      __A							)
		__A :			List[Any]		    =			approximate




	def        lowerCAmelCase_       (     self						:     str					,      __A						:     Optional[Any]							):
		if gate.device.type != "mps":
			return F.gelu(__A					,      approximate=self.approximate							)
		# mps: gelu is not implemented for float16
		return F.gelu(gate.to(dtype=torch.floataa							)					,      approximate=self.approximate							).to(dtype=gate.dtype							)




	def        lowerCAmelCase_       (     self						:     List[Any]					,      __A						:     Optional[int]							):
		__A :			Union[str, Any]		    =			self.proj(__A							)
		__A :			Tuple		    =			self.gelu(__A							)
		return hidden_states






class      lowerCamelCase_      (	nn.Module	):




	def __init__(     self						:     Union[str, Any]					,      __A						:     int					,      __A						:     int							):
		super().__init__()
		__A :			Optional[int]		    =			nn.Linear(__A					,      dim_out * 2							)




	def        lowerCAmelCase_       (     self						:     Tuple					,      __A						:     Tuple							):
		if gate.device.type != "mps":
			return F.gelu(__A							)
		# mps: gelu is not implemented for float16
		return F.gelu(gate.to(dtype=torch.floataa							)							).to(dtype=gate.dtype							)




	def        lowerCAmelCase_       (     self						:     int					,      __A						:     Dict							):
		__A    ,				__A :			Dict		    =			self.proj(__A							).chunk(2					,      dim=-1							)
		return hidden_states * self.gelu(__A							)






class      lowerCamelCase_      (	nn.Module	):




	def __init__(     self						:     Optional[Any]					,      __A						:     int					,      __A						:     int							):
		super().__init__()
		__A :			Tuple		    =			nn.Linear(__A					,      __A							)




	def        lowerCAmelCase_       (     self						:     int					,      __A						:     Tuple							):
		__A :			List[str]		    =			self.proj(__A							)
		return x * torch.sigmoid(1.7_0_2 * x							)






class      lowerCamelCase_      (	nn.Module	):




	def __init__(     self						:     int					,      __A						:     str					,      __A						:     str							):
		super().__init__()
		__A :			Optional[Any]		    =			nn.Embedding(__A					,      __A							)
		__A :			Any		    =			nn.SiLU()
		__A :			Optional[Any]		    =			nn.Linear(__A					,      embedding_dim * 2							)
		__A :			Optional[int]		    =			nn.LayerNorm(__A					,      elementwise_affine=__A							)




	def        lowerCAmelCase_       (     self						:     str					,      __A						:     Any					,      __A						:     Tuple							):
		__A :			List[Any]		    =			self.linear(self.silu(self.emb(__A							)							)							)
		__A    ,				__A :			Union[str, Any]		    =			torch.chunk(__A					,      2							)
		__A :			str		    =			self.norm(__A							) * (1 + scale) + shift
		return x






class      lowerCamelCase_      (	nn.Module	):




	def __init__(     self						:     Tuple					,      __A						:     Union[str, Any]					,      __A						:     int							):
		super().__init__()

		__A :			Any		    =			CombinedTimestepLabelEmbeddings(__A					,      __A							)

		__A :			Any		    =			nn.SiLU()
		__A :			Tuple		    =			nn.Linear(__A					,      6 * embedding_dim					,      bias=__A							)
		__A :			Union[str, Any]		    =			nn.LayerNorm(__A					,      elementwise_affine=__A					,      eps=1e-6							)




	def        lowerCAmelCase_       (     self						:     Tuple					,      __A						:     Any					,      __A						:     Union[str, Any]					,      __A						:     Dict					,      __A						:     Optional[int]=None							):
		__A :			Tuple		    =			self.linear(self.silu(self.emb(__A					,      __A					,      hidden_dtype=__A							)							)							)
		__A    ,				__A    ,				__A    ,				__A    ,				__A    ,				__A :			List[Any]		    =			emb.chunk(6					,      dim=1							)
		__A :			str		    =			self.norm(__A							) * (1 + scale_msa[:, None]) + shift_msa[:, None]
		return x, gate_msa, shift_mlp, scale_mlp, gate_mlp






class      lowerCamelCase_      (	nn.Module	):




	def __init__(     self						:     Dict					,      __A						:     int					,      __A						:     int					,      __A						:     int					,      __A						:     Optional[str] = None					,      __A						:     float = 1e-5							):
		super().__init__()
		__A :			Optional[Any]		    =			num_groups
		__A :			Tuple		    =			eps

		if act_fn is None:
			__A :			Union[str, Any]		    =			None
		else:
			__A :			Tuple		    =			get_activation(__A							)

		__A :			Optional[Any]		    =			nn.Linear(__A					,      out_dim * 2							)




	def        lowerCAmelCase_       (     self						:     List[Any]					,      __A						:     List[Any]					,      __A						:     Optional[int]							):
		if self.act:
			__A :			Union[str, Any]		    =			self.act(__A							)
		__A :			List[Any]		    =			self.linear(__A							)
		__A :			Dict		    =			emb[:, :, None, None]
		__A    ,				__A :			str		    =			emb.chunk(2					,      dim=1							)

		__A :			str		    =			F.group_norm(__A					,      self.num_groups					,      eps=self.eps							)
		__A :			Any		    =			x * (1 + scale) + shift
		return x








import functools
import operator

from ...configuration_utils import PretrainedConfig
from ...utils import logging


UpperCAmelCase_							:					List[str]		 =   logging.get_logger(__name__)

UpperCAmelCase_							:					str		 =   {
    '''microsoft/unispeech-large-1500h-cv''': (
        '''https://huggingface.co/microsoft/unispeech-large-1500h-cv/resolve/main/config.json'''
    ),
    # See all UniSpeech models at https://huggingface.co/models?filter=unispeech
}






class      lowerCamelCase_      (	_lowercase	):
	_lowercase		:   Dict     					=				'''unispeech'''




	def __init__(     self						:     str					,      __A						:     Tuple=32					,      __A						:     List[str]=768					,      __A						:     Dict=12					,      __A						:     Union[str, Any]=12					,      __A						:     Tuple=3072					,      __A						:     Any="gelu"					,      __A						:     int=0.1					,      __A						:     Optional[int]=0.1					,      __A						:     List[Any]=0.1					,      __A						:     Any=0.0					,      __A						:     List[str]=0.0					,      __A						:     int=0.1					,      __A						:     List[Any]=0.1					,      __A						:     List[str]=0.0_2					,      __A						:     List[str]=1e-5					,      __A						:     List[Any]="group"					,      __A						:     int="gelu"					,      __A						:     Any=(512, 512, 512, 512, 512, 512, 512)					,      __A						:     Union[str, Any]=(5, 2, 2, 2, 2, 2, 2)					,      __A						:     Tuple=(10, 3, 3, 3, 3, 2, 2)					,      __A						:     Optional[int]=False					,      __A						:     Any=128					,      __A						:     Union[str, Any]=16					,      __A						:     Optional[Any]=False					,      __A						:     str=True					,      __A						:     Dict=0.0_5					,      __A						:     Optional[Any]=10					,      __A						:     Dict=2					,      __A						:     int=0.0					,      __A						:     List[str]=10					,      __A						:     str=0					,      __A						:     List[str]=320					,      __A						:     List[Any]=2					,      __A						:     Tuple=0.1					,      __A						:     Optional[int]=100					,      __A						:     Any=256					,      __A						:     Dict=256					,      __A						:     Tuple=0.1					,      __A						:     List[str]="mean"					,      __A						:     int=False					,      __A						:     List[str]=False					,      __A						:     List[Any]=256					,      __A						:     str=80					,      __A						:     Tuple=0					,      __A						:     Tuple=1					,      __A						:     int=2					,      __A						:     Dict=0.5					,      **__A						:     List[Any]					,      ):
		super().__init__(**__A					,      pad_token_id=__A					,      bos_token_id=__A					,      eos_token_id=__A							)
		__A :			Dict		    =			hidden_size
		__A :			Optional[Any]		    =			feat_extract_norm
		__A :			List[Any]		    =			feat_extract_activation
		__A :			str		    =			list(__A							)
		__A :			Optional[Any]		    =			list(__A							)
		__A :			Optional[int]		    =			list(__A							)
		__A :			List[Any]		    =			conv_bias
		__A :			Optional[int]		    =			num_conv_pos_embeddings
		__A :			List[Any]		    =			num_conv_pos_embedding_groups
		__A :			int		    =			len(self.conv_dim							)
		__A :			Optional[Any]		    =			num_hidden_layers
		__A :			List[str]		    =			intermediate_size
		__A :			Union[str, Any]		    =			hidden_act
		__A :			Optional[int]		    =			num_attention_heads
		__A :			Tuple		    =			hidden_dropout
		__A :			Optional[Any]		    =			attention_dropout
		__A :			Union[str, Any]		    =			activation_dropout
		__A :			Dict		    =			feat_proj_dropout
		__A :			Optional[int]		    =			final_dropout
		__A :			Dict		    =			layerdrop
		__A :			Optional[int]		    =			layer_norm_eps
		__A :			Optional[Any]		    =			initializer_range
		__A :			Optional[int]		    =			num_ctc_classes
		__A :			Dict		    =			vocab_size
		__A :			List[str]		    =			do_stable_layer_norm
		__A :			Tuple		    =			use_weighted_layer_sum
		__A :			Any		    =			classifier_proj_size

		if (
		    (len(self.conv_stride							) != self.num_feat_extract_layers)
		    or (len(self.conv_kernel							) != self.num_feat_extract_layers)
		    or (len(self.conv_dim							) != self.num_feat_extract_layers)
		):
			raise ValueError(
			    """Configuration for convolutional layers is incorrect. It is required that `len(config.conv_dim)` =="""
			    """ `len(config.conv_stride)` == `len(config.conv_kernel)`, but is `len(config.conv_dim) ="""
			    F""" {len(self.conv_dim							)}`, `len(config.conv_stride) = {len(self.conv_stride							)}`,"""
			    F""" `len(config.conv_kernel) = {len(self.conv_kernel							)}`."""							)

		# fine-tuning config parameters for SpecAugment: https://arxiv.org/abs/1904.08779
		__A :			Tuple		    =			apply_spec_augment
		__A :			Union[str, Any]		    =			mask_time_prob
		__A :			Optional[Any]		    =			mask_time_length
		__A :			List[Any]		    =			mask_time_min_masks
		__A :			List[Any]		    =			mask_feature_prob
		__A :			Any		    =			mask_feature_length
		__A :			List[Any]		    =			mask_feature_min_masks

		# parameters for pretraining with codevector quantized representations
		__A :			Any		    =			num_codevectors_per_group
		__A :			Tuple		    =			num_codevector_groups
		__A :			List[str]		    =			contrastive_logits_temperature
		__A :			Optional[int]		    =			feat_quantizer_dropout
		__A :			int		    =			num_negatives
		__A :			List[str]		    =			codevector_dim
		__A :			int		    =			proj_codevector_dim
		__A :			Union[str, Any]		    =			diversity_loss_weight

		# ctc loss
		__A :			List[str]		    =			ctc_loss_reduction
		__A :			Any		    =			ctc_zero_infinity

		# pretraining loss
		__A :			Union[str, Any]		    =			replace_prob




	@property
	def        lowerCAmelCase_       (     self						:     int							):
		return functools.reduce(operator.mul					,      self.conv_stride					,      1							)





