
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        _UpperCamelCase				:






			'''simple docstring'''

			__UpperCamelCase      :						Dict					      = LEDConfig
			__UpperCamelCase      :						Optional[int]					      = {}
			__UpperCamelCase      :						List[str]					      = """gelu"""

			def __init__(       self :       int						,	snake_case_ :       Optional[int]						,	snake_case_ :       Optional[int]=13						,	snake_case_ :       List[str]=7						,	snake_case_ :       Any=True						,	snake_case_ :       List[str]=False						,	snake_case_ :       Dict=99						,	snake_case_ :       Optional[int]=32						,	snake_case_ :       str=2						,	snake_case_ :       List[Any]=4						,	snake_case_ :       str=37						,	snake_case_ :       Optional[Any]=0.1						,	snake_case_ :       str=0.1						,	snake_case_ :       List[Any]=20						,	snake_case_ :       Tuple=2						,	snake_case_ :       int=1						,	snake_case_ :       Union[str, Any]=0						,	snake_case_ :       List[str]=4						,	):
							UpperCamelCase_:							str	=						parent
							UpperCamelCase_:							Optional[Any]	=						batch_size
							UpperCamelCase_:							Dict	=						seq_length
							UpperCamelCase_:							List[Any]	=						is_training
							UpperCamelCase_:							List[str]	=						use_labels
							UpperCamelCase_:							Tuple	=						vocab_size
							UpperCamelCase_:							str	=						hidden_size
							UpperCamelCase_:							Tuple	=						num_hidden_layers
							UpperCamelCase_:							Dict	=						num_attention_heads
							UpperCamelCase_:							Dict	=						intermediate_size

							UpperCamelCase_:							Dict	=						hidden_dropout_prob
							UpperCamelCase_:							Dict	=						attention_probs_dropout_prob
							UpperCamelCase_:							Dict	=						max_position_embeddings
							UpperCamelCase_:							Union[str, Any]	=						eos_token_id
							UpperCamelCase_:							Union[str, Any]	=						pad_token_id
							UpperCamelCase_:							int	=						bos_token_id
							UpperCamelCase_:							Union[str, 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
							UpperCamelCase_:							Optional[int]	=						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
							UpperCamelCase_:							Dict	=						(
							    self.seq_length + (self.attention_window - self.seq_length % self.attention_window) % self.attention_window
							)

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

							UpperCamelCase_:							List[Any]	=						ids_tensor([self.batch_size, self.seq_length]						,	self.vocab_size      )

							UpperCamelCase_:							str	=						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						,	)
							UpperCamelCase_:							List[str]	=						prepare_led_inputs_dict(snake_case_						,	snake_case_						,	snake_case_      )
							UpperCamelCase_:							List[str]	=						tf.concat(
							    [tf.zeros_like(snake_case_      )[:, :-1], tf.ones_like(snake_case_      )[:, -1:]]						,	axis=-1						,	)
							UpperCamelCase_:							Optional[Any]	=						global_attention_mask
							return config, inputs_dict




			def       lowerCAmelCase__	(       self :       Dict						,	snake_case_ :       List[str]						,	snake_case_ :       int      ):
							UpperCamelCase_:							str	=						TFLEDModel(config=snake_case_      ).get_decoder()
							UpperCamelCase_:							str	=						inputs_dict["""input_ids"""]

							UpperCamelCase_:							Tuple	=						input_ids[:1, :]
							UpperCamelCase_:							Optional[Any]	=						inputs_dict["""attention_mask"""][:1, :]
							UpperCamelCase_:							Dict	=						1

							# first forward pass
							UpperCamelCase_:							List[str]	=						model(snake_case_						,	attention_mask=snake_case_						,	use_cache=snake_case_      )

							UpperCamelCase_,   UpperCamelCase_:							str	=						outputs.to_tuple()

							# create hypothetical next token and extent to next_input_ids
							UpperCamelCase_:							Tuple	=						ids_tensor((self.batch_size, 3)						,	config.vocab_size      )
							UpperCamelCase_:							Tuple	=						tf.cast(ids_tensor((self.batch_size, 3)						,	2      )						,	tf.inta      )

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

							UpperCamelCase_:							int	=						model(snake_case_						,	attention_mask=snake_case_      )[0]
							UpperCamelCase_:							List[Any]	=						model(snake_case_						,	attention_mask=snake_case_						,	past_key_values=snake_case_      )[0]

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

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

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







def        A__					(						lowerCamelCase     ,    lowerCamelCase     ,    lowerCamelCase     ,    lowerCamelCase=None     ,    lowerCamelCase=None     ,    lowerCamelCase=None     ,    lowerCamelCase=None     ,    )  ->	List[Any]:
				if attention_mask is None:
								UpperCamelCase_:							Any	=						tf.cast(tf.math.not_equal(lowerCamelCase     ,    config.pad_token_id						)     ,    tf.inta						)
				if decoder_attention_mask is None:
								UpperCamelCase_:							Optional[Any]	=						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:
								UpperCamelCase_:							Union[str, Any]	=						tf.ones((config.encoder_layers, config.encoder_attention_heads)						)
				if decoder_head_mask is None:
								UpperCamelCase_:							Dict	=						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        _UpperCamelCase				(					_A     ,       _A     ,       unittest.TestCase  ):






			'''simple docstring'''

			__UpperCamelCase      :						Any					      = (TFLEDForConditionalGeneration, TFLEDModel) if is_tf_available() else ()
			__UpperCamelCase      :						Optional[Any]					      = (TFLEDForConditionalGeneration,) if is_tf_available() else ()
			__UpperCamelCase      :						str					      = (
			    {
			        """conversational""": TFLEDForConditionalGeneration,
			        """feature-extraction""": TFLEDModel,
			        """summarization""": TFLEDForConditionalGeneration,
			        """text2text-generation""": TFLEDForConditionalGeneration,
			        """translation""": TFLEDForConditionalGeneration,
			    }
			    if is_tf_available()
			    else {}
			)
			__UpperCamelCase      :						Any					      = True
			__UpperCamelCase      :						List[str]					      = False
			__UpperCamelCase      :						str					      = False
			__UpperCamelCase      :						int					      = False

			def       lowerCAmelCase__	(       self :       List[Any]      ):
							UpperCamelCase_:							List[str]	=						TFLEDModelTester(self      )
							UpperCamelCase_:							Tuple	=						ConfigTester(self						,	config_class=snake_case_      )

			def       lowerCAmelCase__	(       self :       Tuple      ):
							self.config_tester.run_common_tests()

			def       lowerCAmelCase__	(       self :       Any      ):
							UpperCamelCase_:							List[str]	=						self.model_tester.prepare_config_and_inputs_for_common()
							self.model_tester.check_decoder_model_past_large_inputs(*snake_case_      )

			def       lowerCAmelCase__	(       self :       Optional[int]      ):
							UpperCamelCase_,   UpperCamelCase_:							int	=						self.model_tester.prepare_config_and_inputs_for_common()
							UpperCamelCase_:							Tuple	=						tf.zeros_like(inputs_dict["""attention_mask"""]      )
							UpperCamelCase_:							int	=						2
							UpperCamelCase_:							List[Any]	=						tf.where(
							    tf.range(self.model_tester.seq_length      )[None, :] < num_global_attn_indices						,	1						,	inputs_dict["""global_attention_mask"""]						,	)

							UpperCamelCase_:							Any	=						True
							UpperCamelCase_:							Tuple	=						self.model_tester.seq_length
							UpperCamelCase_:							Any	=						self.model_tester.encoder_seq_length

							def check_decoder_attentions_output(snake_case_ :       List[Any]      ):
											UpperCamelCase_:							List[str]	=						outputs.decoder_attentions
											self.assertEqual(len(snake_case_      )						,	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(snake_case_ :       Dict      ):
											UpperCamelCase_:							Dict	=						[t.numpy() for t in outputs.encoder_attentions]
											UpperCamelCase_:							Tuple	=						[t.numpy() for t in outputs.encoder_global_attentions]
											self.assertEqual(len(snake_case_      )						,	self.model_tester.num_hidden_layers      )
											self.assertEqual(len(snake_case_      )						,	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:
											UpperCamelCase_:							Dict	=						True
											UpperCamelCase_:							Dict	=						False
											UpperCamelCase_:							Tuple	=						False
											UpperCamelCase_:							List[str]	=						model_class(snake_case_      )
											UpperCamelCase_:							List[Any]	=						model(self._prepare_for_class(snake_case_						,	snake_case_      )      )
											UpperCamelCase_:							str	=						len(snake_case_      )
											self.assertEqual(config.output_hidden_states						,	snake_case_      )
											check_encoder_attentions_output(snake_case_      )

											if self.is_encoder_decoder:
															UpperCamelCase_:							List[Any]	=						model_class(snake_case_      )
															UpperCamelCase_:							List[str]	=						model(self._prepare_for_class(snake_case_						,	snake_case_      )      )
															self.assertEqual(config.output_hidden_states						,	snake_case_      )
															check_decoder_attentions_output(snake_case_      )

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

											# Check attention is always last and order is fine
											UpperCamelCase_:							str	=						True
											UpperCamelCase_:							Tuple	=						True
											UpperCamelCase_:							Any	=						model_class(snake_case_      )
											UpperCamelCase_:							Dict	=						model(self._prepare_for_class(snake_case_						,	snake_case_      )      )

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

			@unittest.skip("""LED keeps using potentially symbolic tensors in conditionals and breaks tracing."""      )
			def       lowerCAmelCase__	(       self :       Tuple      ):
							pass




			def       lowerCAmelCase__	(       self :       List[Any]      ):
							# TODO: Head-masking not yet implement
							pass







def        A__					(						lowerCamelCase						)  ->	List[Any]:
				return tf.constant(lowerCamelCase     ,    dtype=tf.intaa						)


lowerCamelCase_						: Dict         =  1E-4





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






			'''simple docstring'''

			def       lowerCAmelCase__	(       self :       Union[str, Any]      ):
							UpperCamelCase_:							Union[str, Any]	=						TFLEDForConditionalGeneration.from_pretrained("""allenai/led-base-16384"""      ).led

							# change to intended input here
							UpperCamelCase_:							Dict	=						_long_tensor([512 * [0, 3_1414, 232, 328, 740, 1140, 1_2695, 69]]      )
							UpperCamelCase_:							int	=						_long_tensor([128 * [0, 3_1414, 232, 328, 740, 1140, 1_2695, 69]]      )
							UpperCamelCase_:							Dict	=						prepare_led_inputs_dict(model.config						,	snake_case_						,	snake_case_      )
							UpperCamelCase_:							Optional[Any]	=						model(**snake_case_      )[0]
							UpperCamelCase_:							List[str]	=						(1, 1024, 768)
							self.assertEqual(output.shape						,	snake_case_      )
							# change to expected output here
							UpperCamelCase_:							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]						,	snake_case_						,	atol=1e-3      )




			def       lowerCAmelCase__	(       self :       Tuple      ):
							UpperCamelCase_:							Union[str, Any]	=						TFLEDForConditionalGeneration.from_pretrained("""allenai/led-base-16384"""      )

							# change to intended input here
							UpperCamelCase_:							Union[str, Any]	=						_long_tensor([512 * [0, 3_1414, 232, 328, 740, 1140, 1_2695, 69]]      )
							UpperCamelCase_:							Optional[int]	=						_long_tensor([128 * [0, 3_1414, 232, 328, 740, 1140, 1_2695, 69]]      )
							UpperCamelCase_:							str	=						prepare_led_inputs_dict(model.config						,	snake_case_						,	snake_case_      )
							UpperCamelCase_:							Optional[int]	=						model(**snake_case_      )[0]
							UpperCamelCase_:							Optional[Any]	=						(1, 1024, model.config.vocab_size)
							self.assertEqual(output.shape						,	snake_case_      )
							# change to expected output here
							UpperCamelCase_:							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]						,	snake_case_						,	atol=1e-3						,	rtol=1e-3      )








import json
import logging
import os
import sys
from pathlib import Path

import finetune_rag

from transformers.file_utils import is_apex_available
from transformers.testing_utils import (
    TestCasePlus,
    execute_subprocess_async,
    require_ray,
    require_torch_gpu,
    require_torch_multi_gpu,
)


logging.basicConfig(level=logging.DEBUG)
lowerCamelCase_						: Union[str, Any]         =  logging.getLogger()

lowerCamelCase_						: List[str]         =  logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)

class        _UpperCamelCase				(					_A  ):






			'''simple docstring'''

			def       lowerCAmelCase__	(       self :       List[Any]						,	snake_case_ :       Dict      ):
							os.makedirs(snake_case_						,	exist_ok=snake_case_      )
							UpperCamelCase_:							int	=						{"""source""": """What is love ?""", """target""": """life"""}
							UpperCamelCase_:							Tuple	=						{"""train""": 12, """val""": 2, """test""": 2}
							for split in ["train", "test", "val"]:
											for field in ["source", "target"]:
															UpperCamelCase_:							Tuple	=						"""\n""".join([contents[field]] * n_lines[split]      )
															with open(os.path.join(snake_case_						,	f'''{split}.{field}'''      )						,	"""w"""      ) as f:
																			f.write(snake_case_      )

			def       lowerCAmelCase__	(       self :       Dict						,	snake_case_ :       int						,	snake_case_ :       str = "pytorch"      ):
							UpperCamelCase_:							Optional[Any]	=						self.get_auto_remove_tmp_dir()
							UpperCamelCase_:							Dict	=						os.path.join(snake_case_						,	"""output"""      )
							UpperCamelCase_:							Any	=						os.path.join(snake_case_						,	"""data"""      )
							self._create_dummy_data(data_dir=snake_case_      )

							UpperCamelCase_:							Union[str, Any]	=						f'''
                --data_dir {data_dir} \
                --output_dir {output_dir} \
                --model_name_or_path facebook/rag-sequence-base \
                --model_type rag_sequence \
                --do_train \
                --do_predict \
                --n_val -1 \
                --val_check_interval 1.0 \
                --train_batch_size 2 \
                --eval_batch_size 1 \
                --max_source_length 25 \
                --max_target_length 25 \
                --val_max_target_length 25 \
                --test_max_target_length 25 \
                --label_smoothing 0.1 \
                --dropout 0.1 \
                --attention_dropout 0.1 \
                --weight_decay 0.001 \
                --adam_epsilon 1e-08 \
                --max_grad_norm 0.1 \
                --lr_scheduler polynomial \
                --learning_rate 3e-04 \
                --num_train_epochs 1 \
                --warmup_steps 4 \
                --gradient_accumulation_steps 1 \
                --distributed-port 8787 \
                --use_dummy_dataset 1 \
                --distributed_retriever {distributed_retriever} \
            '''.split()

							if gpus > 0:
											testargs.append(f'''--gpus={gpus}'''      )
											if is_apex_available():
															testargs.append("""--fp16"""      )
							else:
											testargs.append("""--gpus=0"""      )
											testargs.append("""--distributed_backend=ddp_cpu"""      )
											testargs.append("""--num_processes=2"""      )

							UpperCamelCase_:							Optional[Any]	=						[sys.executable, str(Path(finetune_rag.__file__      ).resolve()      )] + testargs
							execute_subprocess_async(snake_case_						,	env=self.get_env()      )

							UpperCamelCase_:							Optional[int]	=						os.path.join(snake_case_						,	"""metrics.json"""      )
							with open(snake_case_      ) as f:
											UpperCamelCase_:							Any	=						json.load(snake_case_      )
							return result

			@require_torch_gpu
			def       lowerCAmelCase__	(       self :       Optional[Any]      ):
							UpperCamelCase_:							List[str]	=						self._run_finetune(gpus=1      )
							self.assertGreaterEqual(result["""test"""][0]["""test_avg_em"""]						,	0.2      )

			@require_torch_multi_gpu
			def       lowerCAmelCase__	(       self :       Optional[Any]      ):
							UpperCamelCase_:							Any	=						self._run_finetune(gpus=2      )
							self.assertGreaterEqual(result["""test"""][0]["""test_avg_em"""]						,	0.2      )

			@require_torch_gpu
			@require_ray
			def       lowerCAmelCase__	(       self :       Dict      ):
							UpperCamelCase_:							List[Any]	=						self._run_finetune(gpus=1						,	distributed_retriever="""ray"""      )
							self.assertGreaterEqual(result["""test"""][0]["""test_avg_em"""]						,	0.2      )




			@require_torch_multi_gpu
			@require_ray
			def       lowerCAmelCase__	(       self :       Dict      ):
							UpperCamelCase_:							List[Any]	=						self._run_finetune(gpus=1						,	distributed_retriever="""ray"""      )
							self.assertGreaterEqual(result["""test"""][0]["""test_avg_em"""]						,	0.2      )






