






import math

import torch
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from .attention_processor import Attention
from .embeddings import get_timestep_embedding
from .modeling_utils import ModelMixin



class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_			,       UpperCamelCase_						):





					"""simple docstring"""






					@register_to_config
					def __init__( self					:  Optional[Any]							,		__A					:  int = 1_2_8							,		__A					:  int = 2_5_6							,		__A					:  float = 2_0_0_0.0							,		__A					:  int = 7_6_8							,		__A					:  int = 1_2							,		__A					:  int = 1_2							,		__A					:  int = 6_4							,		__A					:  int = 2_0_4_8							,		__A					:  float = 0.1							,		):
						super().__init__()

						snake_case__			:				Any       =  nn.Sequential(
						    nn.Linear(__A							,		d_model * 4							,		bias=__A	)							,		nn.SiLU()							,		nn.Linear(d_model * 4							,		d_model * 4							,		bias=__A	)							,		nn.SiLU()							,		)

						snake_case__			:				int       =  nn.Embedding(__A							,		__A	)
						snake_case__			:				List[str]       =  False

						snake_case__			:				List[str]       =  nn.Linear(__A							,		__A							,		bias=__A	)

						snake_case__			:				Dict       =  nn.Dropout(p=__A	)

						snake_case__			:				int       =  nn.ModuleList()
						for lyr_num in range(__A	):
							# FiLM conditional T5 decoder
							snake_case__			:				Any       =  DecoderLayer(d_model=__A							,		d_kv=__A							,		num_heads=__A							,		d_ff=__A							,		dropout_rate=__A	)
							self.decoders.append(__A	)

						snake_case__			:				str       =  TaLayerNorm(__A	)

						snake_case__			:				Union[str, Any]       =  nn.Dropout(p=__A	)
						snake_case__			:				Optional[int]       =  nn.Linear(__A							,		__A							,		bias=__A	)



					def 	_lowercase					( self					:  Tuple							,		__A					:  Tuple							,		__A					:  Optional[Any]	):
						snake_case__			:				List[str]       =  torch.mul(query_input.unsqueeze(-1	)							,		key_input.unsqueeze(-2	)	)
						return mask.unsqueeze(-3	)






					def 	_lowercase					( self					:  List[str]							,		__A					:  Tuple							,		__A					:  Union[str, Any]							,		__A					:  Dict	):
						snake_case__,						snake_case__,						snake_case__			:				Optional[int]       =  decoder_input_tokens.shape
						assert decoder_noise_time.shape == (batch,)

						# decoder_noise_time is in [0, 1), so rescale to expected timing range.
						snake_case__			:				Tuple       =  get_timestep_embedding(
						    decoder_noise_time * self.config.max_decoder_noise_time							,		embedding_dim=self.config.d_model							,		max_period=self.config.max_decoder_noise_time							,		).to(dtype=self.dtype	)

						snake_case__			:				int       =  self.conditioning_emb(__A	).unsqueeze(1	)

						assert conditioning_emb.shape == (batch, 1, self.config.d_model * 4)

						snake_case__			:				Any       =  decoder_input_tokens.shape[1]

						# If we want to use relative positions for audio context, we can just offset
						# this sequence by the length of encodings_and_masks.
						snake_case__			:				Dict       =  torch.broadcast_to(
						    torch.arange(__A							,		device=decoder_input_tokens.device	)							,		(batch, seq_length)							,		)

						snake_case__			:				Union[str, Any]       =  self.position_encoding(__A	)

						snake_case__			:				Dict       =  self.continuous_inputs_projection(__A	)
						inputs += position_encodings
						snake_case__			:				List[Any]       =  self.dropout(__A	)

						# decoder: No padding present.
						snake_case__			:				Tuple       =  torch.ones(
						    decoder_input_tokens.shape[:2]							,		device=decoder_input_tokens.device							,		dtype=inputs.dtype	)

						# Translate encoding masks to encoder-decoder masks.
						snake_case__			:				List[str]       =  [(x, self.encoder_decoder_mask(__A							,		__A	)) for x, y in encodings_and_masks]

						# cross attend style: concat encodings
						snake_case__			:				Optional[int]       =  torch.cat([x[0] for x in encodings_and_encdec_masks]							,		dim=1	)
						snake_case__			:				List[str]       =  torch.cat([x[1] for x in encodings_and_encdec_masks]							,		dim=-1	)

						for lyr in self.decoders:
							snake_case__			:				Optional[Any]       =  lyr(
							    __A							,		conditioning_emb=__A							,		encoder_hidden_states=__A							,		encoder_attention_mask=__A							,		)[0]

						snake_case__			:				int       =  self.decoder_norm(__A	)
						snake_case__			:				List[str]       =  self.post_dropout(__A	)

						snake_case__			:				List[Any]       =  self.spec_out(__A	)
						return spec_out



class 							SCREAMING_SNAKE_CASE__						(    nn.Module						):





					"""simple docstring"""






					def __init__( self					:  Optional[int]							,		__A					:  Dict							,		__A					:  Optional[int]							,		__A					:  Tuple							,		__A					:  Dict							,		__A					:  Optional[int]							,		__A					:  List[str]=1e-6	):
						super().__init__()
						snake_case__			:				Optional[Any]       =  nn.ModuleList()

						# cond self attention: layer 0
						self.layer.append(
						    TaLayerSelfAttentionCond(d_model=__A							,		d_kv=__A							,		num_heads=__A							,		dropout_rate=__A	)	)

						# cross attention: layer 1
						self.layer.append(
						    TaLayerCrossAttention(
						        d_model=__A							,		d_kv=__A							,		num_heads=__A							,		dropout_rate=__A							,		layer_norm_epsilon=__A							,		)	)

						# Film Cond MLP + dropout: last layer
						self.layer.append(
						    TaLayerFFCond(d_model=__A							,		d_ff=__A							,		dropout_rate=__A							,		layer_norm_epsilon=__A	)	)






					def 	_lowercase					( self					:  Optional[int]							,		__A					:  Dict							,		__A					:  int=None							,		__A					:  List[str]=None							,		__A					:  str=None							,		__A					:  Any=None							,		__A					:  str=None							,		):
						snake_case__			:				List[Any]       =  self.layer[0](
						    __A							,		conditioning_emb=__A							,		attention_mask=__A							,		)

						if encoder_hidden_states is not None:
							snake_case__			:				Optional[int]       =  torch.where(encoder_attention_mask > 0							,		0							,		-1e1_0	).to(
							    encoder_hidden_states.dtype	)

							snake_case__			:				Any       =  self.layer[1](
							    __A							,		key_value_states=__A							,		attention_mask=__A							,		)

						# Apply Film Conditional Feed Forward layer
						snake_case__			:				Tuple       =  self.layer[-1](__A							,		__A	)

						return (hidden_states,)



class 							SCREAMING_SNAKE_CASE__						(    nn.Module						):





					"""simple docstring"""






					def __init__( self					:  Tuple							,		__A					:  Any							,		__A					:  List[str]							,		__A					:  Optional[int]							,		__A					:  Any	):
						super().__init__()
						snake_case__			:				Optional[int]       =  TaLayerNorm(__A	)
						snake_case__			:				List[Any]       =  TaFiLMLayer(in_features=d_model * 4							,		out_features=__A	)
						snake_case__			:				Optional[Any]       =  Attention(query_dim=__A							,		heads=__A							,		dim_head=__A							,		out_bias=__A							,		scale_qk=__A	)
						snake_case__			:				List[Any]       =  nn.Dropout(__A	)






					def 	_lowercase					( self					:  str							,		__A					:  Tuple							,		__A					:  Union[str, Any]=None							,		__A					:  Union[str, Any]=None							,		):
						# pre_self_attention_layer_norm
						snake_case__			:				Optional[Any]       =  self.layer_norm(__A	)

						if conditioning_emb is not None:
							snake_case__			:				Optional[Any]       =  self.FiLMLayer(__A							,		__A	)

						# Self-attention block
						snake_case__			:				Union[str, Any]       =  self.attention(__A	)

						snake_case__			:				Any       =  hidden_states + self.dropout(__A	)

						return hidden_states



class 							SCREAMING_SNAKE_CASE__						(    nn.Module						):





					"""simple docstring"""






					def __init__( self					:  Dict							,		__A					:  Any							,		__A					:  List[str]							,		__A					:  List[str]							,		__A					:  Optional[Any]							,		__A					:  int	):
						super().__init__()
						snake_case__			:				Tuple       =  Attention(query_dim=__A							,		heads=__A							,		dim_head=__A							,		out_bias=__A							,		scale_qk=__A	)
						snake_case__			:				Optional[int]       =  TaLayerNorm(__A							,		eps=__A	)
						snake_case__			:				str       =  nn.Dropout(__A	)






					def 	_lowercase					( self					:  Union[str, Any]							,		__A					:  str							,		__A					:  Union[str, Any]=None							,		__A					:  List[str]=None							,		):
						snake_case__			:				Union[str, Any]       =  self.layer_norm(__A	)
						snake_case__			:				Union[str, Any]       =  self.attention(
						    __A							,		encoder_hidden_states=__A							,		attention_mask=attention_mask.squeeze(1	)							,		)
						snake_case__			:				Optional[Any]       =  hidden_states + self.dropout(__A	)
						return layer_output



class 							SCREAMING_SNAKE_CASE__						(    nn.Module						):





					"""simple docstring"""






					def __init__( self					:  Optional[int]							,		__A					:  Optional[int]							,		__A					:  List[Any]							,		__A					:  List[Any]							,		__A					:  Optional[int]	):
						super().__init__()
						snake_case__			:				Union[str, Any]       =  TaDenseGatedActDense(d_model=__A							,		d_ff=__A							,		dropout_rate=__A	)
						snake_case__			:				Tuple       =  TaFiLMLayer(in_features=d_model * 4							,		out_features=__A	)
						snake_case__			:				List[Any]       =  TaLayerNorm(__A							,		eps=__A	)
						snake_case__			:				Tuple       =  nn.Dropout(__A	)






					def 	_lowercase					( self					:  int							,		__A					:  int							,		__A					:  Any=None	):
						snake_case__			:				str       =  self.layer_norm(__A	)
						if conditioning_emb is not None:
							snake_case__			:				int       =  self.film(__A							,		__A	)

						snake_case__			:				List[Any]       =  self.DenseReluDense(__A	)
						snake_case__			:				Dict       =  hidden_states + self.dropout(__A	)
						return hidden_states



class 							SCREAMING_SNAKE_CASE__						(    nn.Module						):





					"""simple docstring"""






					def __init__( self					:  int							,		__A					:  Dict							,		__A					:  str							,		__A					:  Tuple	):
						super().__init__()
						snake_case__			:				List[str]       =  nn.Linear(__A							,		__A							,		bias=__A	)
						snake_case__			:				int       =  nn.Linear(__A							,		__A							,		bias=__A	)
						snake_case__			:				Union[str, Any]       =  nn.Linear(__A							,		__A							,		bias=__A	)
						snake_case__			:				Union[str, Any]       =  nn.Dropout(__A	)
						snake_case__			:				List[Any]       =  NewGELUActivation()






					def 	_lowercase					( self					:  List[Any]							,		__A					:  Any	):
						snake_case__			:				Dict       =  self.act(self.wi_a(__A	)	)
						snake_case__			:				Union[str, Any]       =  self.wi_a(__A	)
						snake_case__			:				Union[str, Any]       =  hidden_gelu * hidden_linear
						snake_case__			:				Union[str, Any]       =  self.dropout(__A	)

						snake_case__			:				Optional[int]       =  self.wo(__A	)
						return hidden_states



class 							SCREAMING_SNAKE_CASE__						(    nn.Module						):





					"""simple docstring"""






					def __init__( self					:  str							,		__A					:  Dict							,		__A					:  Dict=1e-6	):
						super().__init__()
						snake_case__			:				List[Any]       =  nn.Parameter(torch.ones(__A	)	)
						snake_case__			:				List[str]       =  eps






					def 	_lowercase					( self					:  Any							,		__A					:  Union[str, Any]	):
						# T5 uses a layer_norm which only scales and doesn't shift, which is also known as Root Mean
						# Square Layer Normalization https://arxiv.org/abs/1910.07467 thus variance is calculated
						# w/o mean and there is no bias. Additionally we want to make sure that the accumulation for
						# half-precision inputs is done in fp32

						snake_case__			:				Union[str, Any]       =  hidden_states.to(torch.floataa	).pow(2	).mean(-1							,		keepdim=__A	)
						snake_case__			:				List[Any]       =  hidden_states * torch.rsqrt(variance + self.variance_epsilon	)

						# convert into half-precision if necessary
						if self.weight.dtype in [torch.floataa, torch.bfloataa]:
							snake_case__			:				Optional[Any]       =  hidden_states.to(self.weight.dtype	)

						return self.weight * hidden_states



class 							SCREAMING_SNAKE_CASE__						(    nn.Module						):





					"""simple docstring"""






					def 	_lowercase					( self					:  int							,		__A					:  torch.Tensor	):
						return 0.5 * input * (1.0 + torch.tanh(math.sqrt(2.0 / math.pi	) * (input + 0.0_4_4_7_1_5 * torch.pow(__A							,		3.0	))	))



class 							SCREAMING_SNAKE_CASE__						(    nn.Module						):





					"""simple docstring"""






					def __init__( self					:  List[str]							,		__A					:  Dict							,		__A					:  Optional[int]	):
						super().__init__()
						snake_case__			:				Tuple       =  nn.Linear(__A							,		out_features * 2							,		bias=__A	)






					def 	_lowercase					( self					:  Optional[int]							,		__A					:  Any							,		__A					:  List[str]	):
						snake_case__			:				List[str]       =  self.scale_bias(__A	)
						snake_case__,						snake_case__			:				Optional[Any]       =  torch.chunk(__A							,		2							,		-1	)
						snake_case__			:				str       =  x * (1 + scale) + shift
						return x











from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


__lowerCamelCase :      Union[str, Any]										=				{
    """configuration_time_series_transformer""": [
        """TIME_SERIES_TRANSFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP""",
        """TimeSeriesTransformerConfig""",
    ],
}

try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__lowerCamelCase :      Tuple										=				[
	    """TIME_SERIES_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST""",
	    """TimeSeriesTransformerForPrediction""",
	    """TimeSeriesTransformerModel""",
	    """TimeSeriesTransformerPreTrainedModel""",
	]


if TYPE_CHECKING:
	from .configuration_time_series_transformer import (
	    TIME_SERIES_TRANSFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP,
	    TimeSeriesTransformerConfig,
	)

	try:
		if not is_torch_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_time_series_transformer import (
		    TIME_SERIES_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
		    TimeSeriesTransformerForPrediction,
		    TimeSeriesTransformerModel,
		    TimeSeriesTransformerPreTrainedModel,
		)

else:
	import sys

	__lowerCamelCase :      Tuple										=				_LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)



