






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 							snake_case_       (    __UpperCamelCase		,							__UpperCamelCase      ):

						"""simple docstring"""



						@register_to_config
						def __init__(self:					int ,					__UpperCAmelCase:					int = 128 ,					__UpperCAmelCase:					int = 256 ,					__UpperCAmelCase:					float = 20_00.0 ,					__UpperCAmelCase:					int = 768 ,					__UpperCAmelCase:					int = 12 ,					__UpperCAmelCase:					int = 12 ,					__UpperCAmelCase:					int = 64 ,					__UpperCAmelCase:					int = 2048 ,					__UpperCAmelCase:					float = 0.1 ,					)    ->     Dict:




											'''simple docstring'''



											super().__init__()

											__a					:						Optional[Any]      =					nn.Sequential(
											    nn.Linear(__UpperCAmelCase ,					d_model * 4 ,					bias=__UpperCAmelCase   ) ,					nn.SiLU() ,					nn.Linear(d_model * 4 ,					d_model * 4 ,					bias=__UpperCAmelCase   ) ,					nn.SiLU() ,					)

											__a					:						Optional[Any]      =					nn.Embedding(__UpperCAmelCase ,					__UpperCAmelCase   )
											__a					:						int      =					False

											__a					:						str      =					nn.Linear(__UpperCAmelCase ,					__UpperCAmelCase ,					bias=__UpperCAmelCase   )

											__a					:						Optional[int]      =					nn.Dropout(p=__UpperCAmelCase   )

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

											__a					:						Any      =					TaLayerNorm(__UpperCAmelCase   )

											__a					:						Dict      =					nn.Dropout(p=__UpperCAmelCase   )
											__a					:						str      =					nn.Linear(__UpperCAmelCase ,					__UpperCAmelCase ,					bias=__UpperCAmelCase   )



						def        UpperCAmelCase__					(self:					Optional[Any] ,					__UpperCAmelCase:					Optional[int] ,					__UpperCAmelCase:					List[str]   )    ->     List[str]:




											'''simple docstring'''



											__a					:						Tuple      =					torch.mul(query_input.unsqueeze(-1   ) ,					key_input.unsqueeze(-2   )   )
											return mask.unsqueeze(-3   )



						def        UpperCAmelCase__					(self:					Optional[int] ,					__UpperCAmelCase:					int ,					__UpperCAmelCase:					Optional[Any] ,					__UpperCAmelCase:					Any   )    ->     Optional[Any]:




											'''simple docstring'''



											__a      ,			__a      ,			__a					:						Dict      =					decoder_input_tokens.shape
											assert decoder_noise_time.shape == (batch,)

											# decoder_noise_time is in [0, 1), so rescale to expected timing range.
											__a					:						List[Any]      =					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   )

											__a					:						int      =					self.conditioning_emb(__UpperCAmelCase   ).unsqueeze(1   )

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

											__a					:						Optional[int]      =					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.
											__a					:						List[str]      =					torch.broadcast_to(
											    torch.arange(__UpperCAmelCase ,					device=decoder_input_tokens.device   ) ,					(batch, seq_length) ,					)

											__a					:						Optional[Any]      =					self.position_encoding(__UpperCAmelCase   )

											__a					:						int      =					self.continuous_inputs_projection(__UpperCAmelCase   )
											inputs += position_encodings
											__a					:						Any      =					self.dropout(__UpperCAmelCase   )

											# decoder: No padding present.
											__a					:						Union[str, Any]      =					torch.ones(
											    decoder_input_tokens.shape[:2] ,					device=decoder_input_tokens.device ,					dtype=inputs.dtype   )

											# Translate encoding masks to encoder-decoder masks.
											__a					:						Union[str, Any]      =					[(x, self.encoder_decoder_mask(__UpperCAmelCase ,					__UpperCAmelCase   )) for x, y in encodings_and_masks]

											# cross attend style: concat encodings
											__a					:						Union[str, Any]      =					torch.cat([x[0] for x in encodings_and_encdec_masks] ,					dim=1   )
											__a					:						str      =					torch.cat([x[1] for x in encodings_and_encdec_masks] ,					dim=-1   )

											for lyr in self.decoders:
																__a					:						Optional[int]      =					lyr(
																    __UpperCAmelCase ,					conditioning_emb=__UpperCAmelCase ,					encoder_hidden_states=__UpperCAmelCase ,					encoder_attention_mask=__UpperCAmelCase ,					)[0]

											__a					:						int      =					self.decoder_norm(__UpperCAmelCase   )
											__a					:						Any      =					self.post_dropout(__UpperCAmelCase   )

											__a					:						List[str]      =					self.spec_out(__UpperCAmelCase   )
											return spec_out




class 							snake_case_       (    nn.Module      ):

						"""simple docstring"""



						def __init__(self:					List[str] ,					__UpperCAmelCase:					Dict ,					__UpperCAmelCase:					List[Any] ,					__UpperCAmelCase:					str ,					__UpperCAmelCase:					Dict ,					__UpperCAmelCase:					Optional[Any] ,					__UpperCAmelCase:					Dict=1E-6   )    ->     Optional[Any]:




											'''simple docstring'''



											super().__init__()
											__a					:						List[Any]      =					nn.ModuleList()

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

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

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



						def        UpperCAmelCase__					(self:					int ,					__UpperCAmelCase:					int ,					__UpperCAmelCase:					str=None ,					__UpperCAmelCase:					int=None ,					__UpperCAmelCase:					List[Any]=None ,					__UpperCAmelCase:					Any=None ,					__UpperCAmelCase:					str=None ,					)    ->     str:




											'''simple docstring'''



											__a					:						int      =					self.layer[0](
											    __UpperCAmelCase ,					conditioning_emb=__UpperCAmelCase ,					attention_mask=__UpperCAmelCase ,					)

											if encoder_hidden_states is not None:
																__a					:						int      =					torch.where(encoder_attention_mask > 0 ,					0 ,					-1E10   ).to(
																    encoder_hidden_states.dtype   )

																__a					:						Optional[int]      =					self.layer[1](
																    __UpperCAmelCase ,					key_value_states=__UpperCAmelCase ,					attention_mask=__UpperCAmelCase ,					)

											# Apply Film Conditional Feed Forward layer
											__a					:						Dict      =					self.layer[-1](__UpperCAmelCase ,					__UpperCAmelCase   )

											return (hidden_states,)




class 							snake_case_       (    nn.Module      ):

						"""simple docstring"""



						def __init__(self:					Union[str, Any] ,					__UpperCAmelCase:					List[Any] ,					__UpperCAmelCase:					Tuple ,					__UpperCAmelCase:					Tuple ,					__UpperCAmelCase:					List[str]   )    ->     List[Any]:




											'''simple docstring'''



											super().__init__()
											__a					:						Dict      =					TaLayerNorm(__UpperCAmelCase   )
											__a					:						List[str]      =					TaFiLMLayer(in_features=d_model * 4 ,					out_features=__UpperCAmelCase   )
											__a					:						List[str]      =					Attention(query_dim=__UpperCAmelCase ,					heads=__UpperCAmelCase ,					dim_head=__UpperCAmelCase ,					out_bias=__UpperCAmelCase ,					scale_qk=__UpperCAmelCase   )
											__a					:						int      =					nn.Dropout(__UpperCAmelCase   )



						def        UpperCAmelCase__					(self:					Union[str, Any] ,					__UpperCAmelCase:					Optional[int] ,					__UpperCAmelCase:					List[str]=None ,					__UpperCAmelCase:					int=None ,					)    ->     Dict:




											'''simple docstring'''



											__a					:						str      =					self.layer_norm(__UpperCAmelCase   )

											if conditioning_emb is not None:
																__a					:						List[str]      =					self.FiLMLayer(__UpperCAmelCase ,					__UpperCAmelCase   )

											# Self-attention block
											__a					:						int      =					self.attention(__UpperCAmelCase   )

											__a					:						Tuple      =					hidden_states + self.dropout(__UpperCAmelCase   )

											return hidden_states




class 							snake_case_       (    nn.Module      ):

						"""simple docstring"""



						def __init__(self:					Union[str, Any] ,					__UpperCAmelCase:					Dict ,					__UpperCAmelCase:					str ,					__UpperCAmelCase:					Optional[int] ,					__UpperCAmelCase:					str ,					__UpperCAmelCase:					Dict   )    ->     str:




											'''simple docstring'''



											super().__init__()
											__a					:						Optional[int]      =					Attention(query_dim=__UpperCAmelCase ,					heads=__UpperCAmelCase ,					dim_head=__UpperCAmelCase ,					out_bias=__UpperCAmelCase ,					scale_qk=__UpperCAmelCase   )
											__a					:						Union[str, Any]      =					TaLayerNorm(__UpperCAmelCase ,					eps=__UpperCAmelCase   )
											__a					:						int      =					nn.Dropout(__UpperCAmelCase   )



						def        UpperCAmelCase__					(self:					Optional[int] ,					__UpperCAmelCase:					List[Any] ,					__UpperCAmelCase:					Optional[Any]=None ,					__UpperCAmelCase:					Optional[Any]=None ,					)    ->     int:




											'''simple docstring'''



											__a					:						List[str]      =					self.layer_norm(__UpperCAmelCase   )
											__a					:						Union[str, Any]      =					self.attention(
											    __UpperCAmelCase ,					encoder_hidden_states=__UpperCAmelCase ,					attention_mask=attention_mask.squeeze(1   ) ,					)
											__a					:						str      =					hidden_states + self.dropout(__UpperCAmelCase   )
											return layer_output




class 							snake_case_       (    nn.Module      ):

						"""simple docstring"""



						def __init__(self:					int ,					__UpperCAmelCase:					Tuple ,					__UpperCAmelCase:					List[Any] ,					__UpperCAmelCase:					Any ,					__UpperCAmelCase:					Tuple   )    ->     Dict:




											'''simple docstring'''



											super().__init__()
											__a					:						int      =					TaDenseGatedActDense(d_model=__UpperCAmelCase ,					d_ff=__UpperCAmelCase ,					dropout_rate=__UpperCAmelCase   )
											__a					:						str      =					TaFiLMLayer(in_features=d_model * 4 ,					out_features=__UpperCAmelCase   )
											__a					:						List[Any]      =					TaLayerNorm(__UpperCAmelCase ,					eps=__UpperCAmelCase   )
											__a					:						Optional[int]      =					nn.Dropout(__UpperCAmelCase   )



						def        UpperCAmelCase__					(self:					str ,					__UpperCAmelCase:					Optional[Any] ,					__UpperCAmelCase:					List[str]=None   )    ->     int:




											'''simple docstring'''



											__a					:						List[str]      =					self.layer_norm(__UpperCAmelCase   )
											if conditioning_emb is not None:
																__a					:						Optional[int]      =					self.film(__UpperCAmelCase ,					__UpperCAmelCase   )

											__a					:						List[str]      =					self.DenseReluDense(__UpperCAmelCase   )
											__a					:						Dict      =					hidden_states + self.dropout(__UpperCAmelCase   )
											return hidden_states




class 							snake_case_       (    nn.Module      ):

						"""simple docstring"""



						def __init__(self:					Any ,					__UpperCAmelCase:					Optional[int] ,					__UpperCAmelCase:					Union[str, Any] ,					__UpperCAmelCase:					Union[str, Any]   )    ->     List[Any]:




											'''simple docstring'''



											super().__init__()
											__a					:						Dict      =					nn.Linear(__UpperCAmelCase ,					__UpperCAmelCase ,					bias=__UpperCAmelCase   )
											__a					:						List[Any]      =					nn.Linear(__UpperCAmelCase ,					__UpperCAmelCase ,					bias=__UpperCAmelCase   )
											__a					:						List[Any]      =					nn.Linear(__UpperCAmelCase ,					__UpperCAmelCase ,					bias=__UpperCAmelCase   )
											__a					:						Optional[Any]      =					nn.Dropout(__UpperCAmelCase   )
											__a					:						Dict      =					NewGELUActivation()



						def        UpperCAmelCase__					(self:					Any ,					__UpperCAmelCase:					Union[str, Any]   )    ->     int:




											'''simple docstring'''



											__a					:						Optional[int]      =					self.act(self.wi_a(__UpperCAmelCase   )   )
											__a					:						Optional[Any]      =					self.wi_a(__UpperCAmelCase   )
											__a					:						Optional[Any]      =					hidden_gelu * hidden_linear
											__a					:						Optional[int]      =					self.dropout(__UpperCAmelCase   )

											__a					:						Dict      =					self.wo(__UpperCAmelCase   )
											return hidden_states




class 							snake_case_       (    nn.Module      ):

						"""simple docstring"""



						def __init__(self:					Optional[Any] ,					__UpperCAmelCase:					Any ,					__UpperCAmelCase:					Union[str, Any]=1E-6   )    ->     Optional[int]:




											'''simple docstring'''



											super().__init__()
											__a					:						List[Any]      =					nn.Parameter(torch.ones(__UpperCAmelCase   )   )
											__a					:						List[str]      =					eps



						def        UpperCAmelCase__					(self:					List[Any] ,					__UpperCAmelCase:					Union[str, Any]   )    ->     List[str]:




											'''simple docstring'''



											__a					:						Dict      =					hidden_states.to(torch.floataa   ).pow(2   ).mean(-1 ,					keepdim=__UpperCAmelCase   )
											__a					:						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]:
																__a					:						Optional[int]      =					hidden_states.to(self.weight.dtype   )

											return self.weight * hidden_states




class 							snake_case_       (    nn.Module      ):

						"""simple docstring"""



						def        UpperCAmelCase__					(self:					List[str] ,					__UpperCAmelCase:					torch.Tensor   )    ->     torch.Tensor:




											'''simple docstring'''



											return 0.5 * input * (1.0 + torch.tanh(math.sqrt(2.0 / math.pi   ) * (input + 0.04_47_15 * torch.pow(__UpperCAmelCase ,					3.0   ))   ))




class 							snake_case_       (    nn.Module      ):

						"""simple docstring"""



						def __init__(self:					Dict ,					__UpperCAmelCase:					str ,					__UpperCAmelCase:					List[str]   )    ->     Dict:




											'''simple docstring'''



											super().__init__()
											__a					:						Tuple      =					nn.Linear(__UpperCAmelCase ,					out_features * 2 ,					bias=__UpperCAmelCase   )



						def        UpperCAmelCase__					(self:					Tuple ,					__UpperCAmelCase:					Optional[int] ,					__UpperCAmelCase:					Union[str, Any]   )    ->     int:




											'''simple docstring'''



											__a					:						Tuple      =					self.scale_bias(__UpperCAmelCase   )
											__a      ,			__a					:						Union[str, Any]      =					torch.chunk(__UpperCAmelCase ,					2 ,					-1   )
											__a					:						Dict      =					x * (1 + scale) + shift
											return x













import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
			import torch_xla.core.xla_model as xm
			import torch_xla.debug.metrics as met




class 							snake_case_       (    __UpperCamelCase      ):

						"""simple docstring"""



						def __init__(self:					str ,					*__UpperCAmelCase:					Optional[int] ,					__UpperCAmelCase:					Optional[int]=None ,					__UpperCAmelCase:					Dict=None ,					**__UpperCAmelCase:					Tuple   )    ->     Dict:




											'''simple docstring'''



											super().__init__(*__UpperCAmelCase ,					**__UpperCAmelCase   )
											__a					:						str      =					eval_examples
											__a					:						List[Any]      =					post_process_function



						def        UpperCAmelCase__					(self:					Optional[int] ,					__UpperCAmelCase:					Optional[Dataset] = None ,					__UpperCAmelCase:					Optional[int]=None ,					__UpperCAmelCase:					Optional[List[str]] = None ,					__UpperCAmelCase:					str = "eval" ,					**__UpperCAmelCase:					Any ,					)    ->     Dict[str, float]:




											'''simple docstring'''



											__a					:						Optional[int]      =					gen_kwargs.copy()
											__a					:						Union[str, Any]      =					(
											    gen_kwargs["max_length"] if gen_kwargs.get("max_length"   ) is not None else self.args.generation_max_length
											)
											__a					:						Dict      =					(
											    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"   ) is not None else self.args.generation_num_beams
											)
											__a					:						Optional[Any]      =					gen_kwargs

											__a					:						Union[str, Any]      =					self.eval_dataset if eval_dataset is None else eval_dataset
											__a					:						Optional[int]      =					self.get_eval_dataloader(__UpperCAmelCase   )
											__a					:						Tuple      =					self.eval_examples if eval_examples is None else eval_examples

											# Temporarily disable metric computation, we will do it in the loop here.
											__a					:						int      =					self.compute_metrics
											__a					:						Dict      =					None
											__a					:						List[Any]      =					time.time()
											__a					:						str      =					self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
											try:
																__a					:						str      =					eval_loop(
																    __UpperCAmelCase ,					description="Evaluation" ,					prediction_loss_only=True if compute_metrics is None else None ,					ignore_keys=__UpperCAmelCase ,					metric_key_prefix=__UpperCAmelCase ,					)
											finally:
																__a					:						Union[str, Any]      =					compute_metrics
											__a					:						str      =					self.args.eval_batch_size * self.args.world_size
											if f'{metric_key_prefix}_jit_compilation_time' in output.metrics:
																start_time += output.metrics[f'{metric_key_prefix}_jit_compilation_time']
											output.metrics.update(
											    speed_metrics(
											        __UpperCAmelCase ,					__UpperCAmelCase ,					num_samples=output.num_samples ,					num_steps=math.ceil(output.num_samples / total_batch_size   ) ,					)   )

											if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
																# Only the main node write the results by default
																__a					:						Union[str, Any]      =					self.post_process_function(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase   )
																__a					:						Dict      =					self.compute_metrics(__UpperCAmelCase   )

																# Prefix all keys with metric_key_prefix + '_'
																for key in list(metrics.keys()   ):
																					if not key.startswith(f'{metric_key_prefix}_'   ):
																										__a					:						int      =					metrics.pop(__UpperCAmelCase   )

																metrics.update(output.metrics   )
											else:
																__a					:						Optional[int]      =					output.metrics

											if self.args.should_log:
																# Only the main node log the results by default
																self.log(__UpperCAmelCase   )

											if self.args.tpu_metrics_debug or self.args.debug:
																# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
																xm.master_print(met.metrics_report()   )

											__a					:						Tuple      =					self.callback_handler.on_evaluate(self.args ,					self.state ,					self.control ,					__UpperCAmelCase   )
											return metrics



						def        UpperCAmelCase__					(self:					Dict ,					__UpperCAmelCase:					List[str] ,					__UpperCAmelCase:					List[str] ,					__UpperCAmelCase:					Union[str, Any]=None ,					__UpperCAmelCase:					str = "test" ,					**__UpperCAmelCase:					List[Any]   )    ->     Union[str, Any]:




											'''simple docstring'''



											__a					:						Any      =					gen_kwargs.copy()

											__a					:						Tuple      =					self.get_test_dataloader(__UpperCAmelCase   )

											# Temporarily disable metric computation, we will do it in the loop here.
											__a					:						Optional[Any]      =					self.compute_metrics
											__a					:						int      =					None
											__a					:						List[Any]      =					time.time()
											__a					:						Dict      =					self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
											try:
																__a					:						Optional[Any]      =					eval_loop(
																    __UpperCAmelCase ,					description="Prediction" ,					prediction_loss_only=True if compute_metrics is None else None ,					ignore_keys=__UpperCAmelCase ,					metric_key_prefix=__UpperCAmelCase ,					)
											finally:
																__a					:						List[Any]      =					compute_metrics

											__a					:						Optional[Any]      =					self.args.eval_batch_size * self.args.world_size
											if f'{metric_key_prefix}_jit_compilation_time' in output.metrics:
																start_time += output.metrics[f'{metric_key_prefix}_jit_compilation_time']
											output.metrics.update(
											    speed_metrics(
											        __UpperCAmelCase ,					__UpperCAmelCase ,					num_samples=output.num_samples ,					num_steps=math.ceil(output.num_samples / total_batch_size   ) ,					)   )
											if self.post_process_function is None or self.compute_metrics is None:
																return output

											__a					:						Optional[int]      =					self.post_process_function(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase ,					"predict"   )
											__a					:						Dict      =					self.compute_metrics(__UpperCAmelCase   )

											# Prefix all keys with metric_key_prefix + '_'
											for key in list(metrics.keys()   ):
																if not key.startswith(f'{metric_key_prefix}_'   ):
																					__a					:						List[str]      =					metrics.pop(__UpperCAmelCase   )
											metrics.update(output.metrics   )
											return PredictionOutput(predictions=predictions.predictions ,					label_ids=predictions.label_ids ,					metrics=__UpperCAmelCase   )





