

"""simple docstring"""





from __future__ import annotations

import unittest

from transformers import LEDConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
							import tensorflow as tf

							from transformers import TFLEDForConditionalGeneration, TFLEDModel







@require_tf
class 				SCREAMING_SNAKE_CASE					:






					"""simple docstring"""



					lowercase__				 =       LEDConfig
					lowercase__				 =       {}
					lowercase__				 =       "gelu"







					def __init__(       self			:				Any   ,lowercase_			:				Optional[Any]   ,lowercase_			:				Optional[int]=1_3   ,lowercase_			:				List[str]=7   ,lowercase_			:				Any=True   ,lowercase_			:				Tuple=False   ,lowercase_			:				Optional[int]=9_9   ,lowercase_			:				str=3_2   ,lowercase_			:				Dict=2   ,lowercase_			:				Dict=4   ,lowercase_			:				int=3_7   ,lowercase_			:				Optional[int]=0.1   ,lowercase_			:				Tuple=0.1   ,lowercase_			:				str=2_0   ,lowercase_			:				Tuple=2   ,lowercase_			:				Dict=1   ,lowercase_			:				List[Any]=0   ,lowercase_			:				List[str]=4   ,):
						lowerCAmelCase__	:						Tuple  				=	parent
						lowerCAmelCase__	:						int  				=	batch_size
						lowerCAmelCase__	:						str  				=	seq_length
						lowerCAmelCase__	:						List[str]  				=	is_training
						lowerCAmelCase__	:						List[str]  				=	use_labels
						lowerCAmelCase__	:						str  				=	vocab_size
						lowerCAmelCase__	:						Tuple  				=	hidden_size
						lowerCAmelCase__	:						Dict  				=	num_hidden_layers
						lowerCAmelCase__	:						List[str]  				=	num_attention_heads
						lowerCAmelCase__	:						Any  				=	intermediate_size

						lowerCAmelCase__	:						Any  				=	hidden_dropout_prob
						lowerCAmelCase__	:						List[str]  				=	attention_probs_dropout_prob
						lowerCAmelCase__	:						Optional[Any]  				=	max_position_embeddings
						lowerCAmelCase__	:						List[Any]  				=	eos_token_id
						lowerCAmelCase__	:						Optional[int]  				=	pad_token_id
						lowerCAmelCase__	:						Any  				=	bos_token_id
						lowerCAmelCase__	:						Any  				=	attention_window

						# `ModelTesterMixin.test_attention_outputs` is expecting attention tensors to be of size
						# [num_attention_heads, encoder_seq_length, encoder_key_length], but TFLongformerSelfAttention
						# returns attention of shape [num_attention_heads, encoder_seq_length, self.attention_window + 1]
						# because its local attention only attends to `self.attention_window` and one before and one after
						lowerCAmelCase__	:						List[str]  				=	self.attention_window + 2

						# because of padding `encoder_seq_length`, is different from `seq_length`. Relevant for
						# the `test_attention_outputs` and `test_hidden_states_output` tests
						lowerCAmelCase__	:						Optional[Any]  				=	(
						    self.seq_length + (self.attention_window - self.seq_length % self.attention_window) % self.attention_window
						)







					def        __lowerCAmelCase (       self			:				int		):
						lowerCAmelCase__	:						str  				=	ids_tensor([self.batch_size, self.seq_length - 1]   ,self.vocab_size		)
						lowerCAmelCase__	:						Any  				=	tf.expand_dims(tf.constant([self.eos_token_id] * self.batch_size		)   ,1		)
						lowerCAmelCase__	:						Dict  				=	tf.concat([input_ids, eos_tensor]   ,axis=1		)

						lowerCAmelCase__	:						int  				=	ids_tensor([self.batch_size, self.seq_length]   ,self.vocab_size		)

						lowerCAmelCase__	:						int  				=	self.config_cls(
						    vocab_size=self.vocab_size   ,d_model=self.hidden_size   ,encoder_layers=self.num_hidden_layers   ,decoder_layers=self.num_hidden_layers   ,encoder_attention_heads=self.num_attention_heads   ,decoder_attention_heads=self.num_attention_heads   ,encoder_ffn_dim=self.intermediate_size   ,decoder_ffn_dim=self.intermediate_size   ,dropout=self.hidden_dropout_prob   ,attention_dropout=self.attention_probs_dropout_prob   ,max_position_embeddings=self.max_position_embeddings   ,eos_token_ids=[2]   ,bos_token_id=self.bos_token_id   ,pad_token_id=self.pad_token_id   ,decoder_start_token_id=self.pad_token_id   ,attention_window=self.attention_window   ,**self.config_updates   ,)
						lowerCAmelCase__	:						List[Any]  				=	prepare_led_inputs_dict(lowercase_   ,lowercase_   ,lowercase_		)
						lowerCAmelCase__	:						Any  				=	tf.concat(
						    [tf.zeros_like(lowercase_		)[:, :-1], tf.ones_like(lowercase_		)[:, -1:]]   ,axis=-1   ,)
						lowerCAmelCase__	:						Optional[int]  				=	global_attention_mask
						return config, inputs_dict







					def        __lowerCAmelCase (       self			:				Any   ,lowercase_			:				str   ,lowercase_			:				Any		):
						lowerCAmelCase__	:						Optional[int]  				=	TFLEDModel(config=lowercase_		).get_decoder()
						lowerCAmelCase__	:						Tuple  				=	inputs_dict['''input_ids''']

						lowerCAmelCase__	:						Optional[int]  				=	input_ids[:1, :]
						lowerCAmelCase__	:						Optional[Any]  				=	inputs_dict['''attention_mask'''][:1, :]
						lowerCAmelCase__	:						str  				=	1

						# first forward pass
						lowerCAmelCase__	:						List[Any]  				=	model(lowercase_   ,attention_mask=lowercase_   ,use_cache=lowercase_		)

						lowerCAmelCase__				,lowerCAmelCase__	:						Tuple  				=	outputs.to_tuple()

						# create hypothetical next token and extent to next_input_ids
						lowerCAmelCase__	:						int  				=	ids_tensor((self.batch_size, 3)   ,config.vocab_size		)
						lowerCAmelCase__	:						Union[str, Any]  				=	tf.cast(ids_tensor((self.batch_size, 3)   ,2		)   ,tf.inta		)

						# append to next input_ids and
						lowerCAmelCase__	:						List[str]  				=	tf.concat([input_ids, next_tokens]   ,axis=-1		)
						lowerCAmelCase__	:						Optional[int]  				=	tf.concat([attention_mask, next_attn_mask]   ,axis=-1		)

						lowerCAmelCase__	:						int  				=	model(lowercase_   ,attention_mask=lowercase_		)[0]
						lowerCAmelCase__	:						int  				=	model(lowercase_   ,attention_mask=lowercase_   ,past_key_values=lowercase_		)[0]

						self.parent.assertEqual(next_tokens.shape[1]   ,output_from_past.shape[1]		)

						# select random slice
						lowerCAmelCase__	:						Union[str, Any]  				=	int(ids_tensor((1,)   ,output_from_past.shape[-1]		)		)
						lowerCAmelCase__	:						List[Any]  				=	output_from_no_past[:, -3:, random_slice_idx]
						lowerCAmelCase__	:						Any  				=	output_from_past[:, :, random_slice_idx]

						# test that outputs are equal for slice
						tf.debugging.assert_near(lowercase_   ,lowercase_   ,rtol=1E-3		)

def   __SCREAMING_SNAKE_CASE  (      A_						,     A_						,     A_						,     A_=None						,     A_=None						,     A_=None						,     A_=None						,     ):
	if attention_mask is None:
		lowerCAmelCase__	:						Optional[Any]  				=	tf.cast(tf.math.not_equal(A_						,     config.pad_token_id )						,     tf.inta )
	if decoder_attention_mask is None:
		lowerCAmelCase__	:						Optional[int]  				=	tf.concat(
		    [
		        tf.ones(decoder_input_ids[:, :1].shape						,     dtype=tf.inta ),
		        tf.cast(tf.math.not_equal(decoder_input_ids[:, 1:]						,     config.pad_token_id )						,     tf.inta ),
		    ]						,     axis=-1						,     )
	if head_mask is None:
		lowerCAmelCase__	:						int  				=	tf.ones((config.encoder_layers, config.encoder_attention_heads) )
	if decoder_head_mask is None:
		lowerCAmelCase__	:						Optional[Any]  				=	tf.ones((config.decoder_layers, config.decoder_attention_heads) )
	return {
	    "input_ids": input_ids,
	    "attention_mask": attention_mask,
	    "decoder_input_ids": decoder_input_ids,
	    "decoder_attention_mask": decoder_attention_mask,
	    "head_mask": head_mask,
	    "decoder_head_mask": decoder_head_mask,
	}







@require_tf
class 				SCREAMING_SNAKE_CASE					(		a_    ,							a_    ,							unittest.TestCase       ):






					"""simple docstring"""



					lowercase__				 =       (TFLEDForConditionalGeneration, TFLEDModel) if is_tf_available() else ()
					lowercase__				 =       (TFLEDForConditionalGeneration,) if is_tf_available() else ()
					lowercase__				 =       (
					    {
					        "conversational": TFLEDForConditionalGeneration,
					        "feature-extraction": TFLEDModel,
					        "summarization": TFLEDForConditionalGeneration,
					        "text2text-generation": TFLEDForConditionalGeneration,
					        "translation": TFLEDForConditionalGeneration,
					    }
					    if is_tf_available()
					    else {}
					)
					lowercase__				 =       True
					lowercase__				 =       False
					lowercase__				 =       False
					lowercase__				 =       False







					def        __lowerCAmelCase (       self			:				List[str]		):
						lowerCAmelCase__	:						int  				=	TFLEDModelTester(self		)
						lowerCAmelCase__	:						Union[str, Any]  				=	ConfigTester(self   ,config_class=lowercase_		)







					def        __lowerCAmelCase (       self			:				Dict		):
						self.config_tester.run_common_tests()







					def        __lowerCAmelCase (       self			:				str		):
						lowerCAmelCase__	:						Optional[int]  				=	self.model_tester.prepare_config_and_inputs_for_common()
						self.model_tester.check_decoder_model_past_large_inputs(*lowercase_		)







					def        __lowerCAmelCase (       self			:				List[Any]		):
						lowerCAmelCase__				,lowerCAmelCase__	:						Dict  				=	self.model_tester.prepare_config_and_inputs_for_common()
						lowerCAmelCase__	:						Any  				=	tf.zeros_like(inputs_dict['''attention_mask''']		)
						lowerCAmelCase__	:						str  				=	2
						lowerCAmelCase__	:						Dict  				=	tf.where(
						    tf.range(self.model_tester.seq_length		)[None, :] < num_global_attn_indices   ,1   ,inputs_dict['''global_attention_mask''']   ,)

						lowerCAmelCase__	:						Union[str, Any]  				=	True
						lowerCAmelCase__	:						Dict  				=	self.model_tester.seq_length
						lowerCAmelCase__	:						List[Any]  				=	self.model_tester.encoder_seq_length

						def check_decoder_attentions_output(lowercase_			:				List[str]		):
							lowerCAmelCase__	:						str  				=	outputs.decoder_attentions
							self.assertEqual(len(lowercase_		)   ,self.model_tester.num_hidden_layers		)
							self.assertListEqual(
							    list(decoder_attentions[0].shape[-3:]		)   ,[self.model_tester.num_attention_heads, seq_length, seq_length]   ,)

						def check_encoder_attentions_output(lowercase_			:				Optional[Any]		):
							lowerCAmelCase__	:						Optional[Any]  				=	[t.numpy() for t in outputs.encoder_attentions]
							lowerCAmelCase__	:						str  				=	[t.numpy() for t in outputs.encoder_global_attentions]
							self.assertEqual(len(lowercase_		)   ,self.model_tester.num_hidden_layers		)
							self.assertEqual(len(lowercase_		)   ,self.model_tester.num_hidden_layers		)
							self.assertListEqual(
							    list(attentions[0].shape[-3:]		)   ,[self.model_tester.num_attention_heads, seq_length, seq_length]   ,)
							self.assertListEqual(
							    list(global_attentions[0].shape[-3:]		)   ,[self.model_tester.num_attention_heads, encoder_seq_length, num_global_attn_indices]   ,)

						for model_class in self.all_model_classes:
							lowerCAmelCase__	:						Union[str, Any]  				=	True
							lowerCAmelCase__	:						Dict  				=	False
							lowerCAmelCase__	:						Optional[Any]  				=	False
							lowerCAmelCase__	:						Tuple  				=	model_class(lowercase_		)
							lowerCAmelCase__	:						List[str]  				=	model(self._prepare_for_class(lowercase_   ,lowercase_		)		)
							lowerCAmelCase__	:						Dict  				=	len(lowercase_		)
							self.assertEqual(config.output_hidden_states   ,lowercase_		)
							check_encoder_attentions_output(lowercase_		)

							if self.is_encoder_decoder:
								lowerCAmelCase__	:						Tuple  				=	model_class(lowercase_		)
								lowerCAmelCase__	:						Any  				=	model(self._prepare_for_class(lowercase_   ,lowercase_		)		)
								self.assertEqual(config.output_hidden_states   ,lowercase_		)
								check_decoder_attentions_output(lowercase_		)

							# Check that output attentions can also be changed via the config
							del inputs_dict["output_attentions"]
							lowerCAmelCase__	:						List[Any]  				=	True
							lowerCAmelCase__	:						Optional[int]  				=	model_class(lowercase_		)
							lowerCAmelCase__	:						List[Any]  				=	model(self._prepare_for_class(lowercase_   ,lowercase_		)		)
							self.assertEqual(config.output_hidden_states   ,lowercase_		)
							check_encoder_attentions_output(lowercase_		)

							# Check attention is always last and order is fine
							lowerCAmelCase__	:						Any  				=	True
							lowerCAmelCase__	:						Union[str, Any]  				=	True
							lowerCAmelCase__	:						str  				=	model_class(lowercase_		)
							lowerCAmelCase__	:						str  				=	model(self._prepare_for_class(lowercase_   ,lowercase_		)		)

							self.assertEqual(out_len + (2 if self.is_encoder_decoder else 1)   ,len(lowercase_		)		)
							self.assertEqual(model.config.output_hidden_states   ,lowercase_		)
							check_encoder_attentions_output(lowercase_		)







					@unittest.skip('''LED keeps using potentially symbolic tensors in conditionals and breaks tracing.'''		)
					def        __lowerCAmelCase (       self			:				Dict		):
						pass







					def        __lowerCAmelCase (       self			:				Dict		):
						# TODO: Head-masking not yet implement
						pass

def   __SCREAMING_SNAKE_CASE  (      A_ ):
	return tf.constant(A_						,     dtype=tf.intaa )


__UpperCamelCase  :      Any							      =		1e-4







@slow
@require_tf
class 				SCREAMING_SNAKE_CASE					(		unittest.TestCase       ):






					"""simple docstring"""



					def        __lowerCAmelCase (       self			:				Dict		):
						lowerCAmelCase__	:						Tuple  				=	TFLEDForConditionalGeneration.from_pretrained('''allenai/led-base-16384'''		).led

						# change to intended input here
						lowerCAmelCase__	:						Tuple  				=	_long_tensor([5_1_2 * [0, 3_1_4_1_4, 2_3_2, 3_2_8, 7_4_0, 1_1_4_0, 1_2_6_9_5, 6_9]]		)
						lowerCAmelCase__	:						Any  				=	_long_tensor([1_2_8 * [0, 3_1_4_1_4, 2_3_2, 3_2_8, 7_4_0, 1_1_4_0, 1_2_6_9_5, 6_9]]		)
						lowerCAmelCase__	:						str  				=	prepare_led_inputs_dict(model.config   ,lowercase_   ,lowercase_		)
						lowerCAmelCase__	:						str  				=	model(**lowercase_		)[0]
						lowerCAmelCase__	:						str  				=	(1, 1_0_2_4, 7_6_8)
						self.assertEqual(output.shape   ,lowercase_		)
						# change to expected output here
						lowerCAmelCase__	:						List[str]  				=	tf.convert_to_tensor(
						    [[2.3050, 2.8279, 0.6531], [-1.8457, -0.1455, -3.5661], [-1.0186, 0.4586, -2.2043]]   ,)
						tf.debugging.assert_near(output[:, :3, :3]   ,lowercase_   ,atol=1E-3		)







					def        __lowerCAmelCase (       self			:				int		):
						lowerCAmelCase__	:						Any  				=	TFLEDForConditionalGeneration.from_pretrained('''allenai/led-base-16384'''		)

						# change to intended input here
						lowerCAmelCase__	:						List[str]  				=	_long_tensor([5_1_2 * [0, 3_1_4_1_4, 2_3_2, 3_2_8, 7_4_0, 1_1_4_0, 1_2_6_9_5, 6_9]]		)
						lowerCAmelCase__	:						Dict  				=	_long_tensor([1_2_8 * [0, 3_1_4_1_4, 2_3_2, 3_2_8, 7_4_0, 1_1_4_0, 1_2_6_9_5, 6_9]]		)
						lowerCAmelCase__	:						Any  				=	prepare_led_inputs_dict(model.config   ,lowercase_   ,lowercase_		)
						lowerCAmelCase__	:						List[str]  				=	model(**lowercase_		)[0]
						lowerCAmelCase__	:						Any  				=	(1, 1_0_2_4, model.config.vocab_size)
						self.assertEqual(output.shape   ,lowercase_		)
						# change to expected output here
						lowerCAmelCase__	:						int  				=	tf.convert_to_tensor(
						    [[33.6507, 6.4572, 16.8089], [5.8739, -2.4238, 11.2902], [-3.2139, -4.3149, 4.2783]]   ,)
						tf.debugging.assert_near(output[:, :3, :3]   ,lowercase_   ,atol=1E-3   ,rtol=1E-3		)






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



	def __init__(    self						,  *lowercase						,  lowercase=None						,  lowercase=None						,  **lowercase )				->       List[str]:
					super().__init__(*lowercase						,  **lowercase )
					lowerCamelCase_						=			eval_examples
					lowerCamelCase_						=			post_process_function



	def     SCREAMING_SNAKE_CASE_(    self						,  lowercase = None						,  lowercase=None						,  lowercase = None						,  lowercase = "eval"						,  **lowercase						,  )				->       Dict[str, float]:
					lowerCamelCase_						=			gen_kwargs.copy()
					lowerCamelCase_						=			(
					    gen_kwargs["max_length"] if gen_kwargs.get("max_length" ) is not None else self.args.generation_max_length
					)
					lowerCamelCase_						=			(
					    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams" ) is not None else self.args.generation_num_beams
					)
					lowerCamelCase_						=			gen_kwargs

					lowerCamelCase_						=			self.eval_dataset if eval_dataset is None else eval_dataset
					lowerCamelCase_						=			self.get_eval_dataloader(lowercase )
					lowerCamelCase_						=			self.eval_examples if eval_examples is None else eval_examples

					# Temporarily disable metric computation, we will do it in the loop here.
					lowerCamelCase_						=			self.compute_metrics
					lowerCamelCase_						=			None
					lowerCamelCase_						=			time.time()
					lowerCamelCase_						=			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
					try:
									lowerCamelCase_						=			eval_loop(
									    lowercase						,  description="Evaluation"						,  prediction_loss_only=True if compute_metrics is None else None						,  ignore_keys=lowercase						,  metric_key_prefix=lowercase						,  )
					finally:
									lowerCamelCase_						=			compute_metrics
					lowerCamelCase_						=			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(
					        lowercase						,  lowercase						,  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
									lowerCamelCase_						=			self.post_process_function(lowercase						,  lowercase						,  lowercase )
									lowerCamelCase_						=			self.compute_metrics(lowercase )

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

									metrics.update(output.metrics )
					else:
									lowerCamelCase_						=			output.metrics

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

					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() )

					lowerCamelCase_						=			self.callback_handler.on_evaluate(self.args						,  self.state						,  self.control						,  lowercase )
					return metrics



	def     SCREAMING_SNAKE_CASE_(    self						,  lowercase						,  lowercase						,  lowercase=None						,  lowercase = "test"						,  **lowercase )				->       Union[str, Any]:
					lowerCamelCase_						=			gen_kwargs.copy()

					lowerCamelCase_						=			self.get_test_dataloader(lowercase )

					# Temporarily disable metric computation, we will do it in the loop here.
					lowerCamelCase_						=			self.compute_metrics
					lowerCamelCase_						=			None
					lowerCamelCase_						=			time.time()
					lowerCamelCase_						=			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
					try:
									lowerCamelCase_						=			eval_loop(
									    lowercase						,  description="Prediction"						,  prediction_loss_only=True if compute_metrics is None else None						,  ignore_keys=lowercase						,  metric_key_prefix=lowercase						,  )
					finally:
									lowerCamelCase_						=			compute_metrics

					lowerCamelCase_						=			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(
					        lowercase						,  lowercase						,  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

					lowerCamelCase_						=			self.post_process_function(lowercase						,  lowercase						,  lowercase						,  "predict" )
					lowerCamelCase_						=			self.compute_metrics(lowercase )

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


