'''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        __A      :
    a__       :						List[Any]          =  LEDConfig
    a__       :						Optional[int]          =  {}
    a__       :						Tuple          =  """gelu"""



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

         UpperCAmelCase_           =   hidden_dropout_prob
         UpperCAmelCase_           =   attention_probs_dropout_prob
         UpperCAmelCase_           =   max_position_embeddings
         UpperCAmelCase_           =   eos_token_id
         UpperCAmelCase_           =   pad_token_id
         UpperCAmelCase_           =   bos_token_id
         UpperCAmelCase_           =   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_           =   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_           =   (
             self.seq_length + (self.attention_window - self.seq_length % self.attention_window) % self.attention_window
         )



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

         UpperCAmelCase_           =   ids_tensor([self.batch_size, self.seq_length]   ,			self.vocab_size )

         UpperCAmelCase_           =   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_           =   prepare_led_inputs_dict(__a   ,			__a   ,			__a )
         UpperCAmelCase_           =   tf.concat(
             [tf.zeros_like(__a )[:, :-1], tf.ones_like(__a )[:, -1:]]   ,			axis=-1   ,			)
         UpperCAmelCase_           =   global_attention_mask
         return config, inputs_dict






    def 					_lowercase						(self    :    Optional[int]   ,			__a    :    int   ,			__a    :    int ):
         UpperCAmelCase_           =   TFLEDModel(config=__a ).get_decoder()
         UpperCAmelCase_           =   inputs_dict["input_ids"]

         UpperCAmelCase_           =   input_ids[:1, :]
         UpperCAmelCase_           =   inputs_dict["attention_mask"][:1, :]
         UpperCAmelCase_           =   1

         # first forward pass
         UpperCAmelCase_           =   model(__a   ,			attention_mask=__a   ,			use_cache=__a )

         UpperCAmelCase_    , UpperCAmelCase_           =   outputs.to_tuple()

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

         # append to next input_ids and
         UpperCAmelCase_           =   tf.concat([input_ids, next_tokens]   ,			axis=-1 )
         UpperCAmelCase_           =   tf.concat([attention_mask, next_attn_mask]   ,			axis=-1 )

         UpperCAmelCase_           =   model(__a   ,			attention_mask=__a )[0]
         UpperCAmelCase_           =   model(__a   ,			attention_mask=__a   ,			past_key_values=__a )[0]

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

         # select random slice
         UpperCAmelCase_           =   int(ids_tensor((1,)   ,			output_from_past.shape[-1] ) )
         UpperCAmelCase_           =   output_from_no_past[:, -3:, random_slice_idx]
         UpperCAmelCase_           =   output_from_past[:, :, random_slice_idx]

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





def 	lowerCAmelCase_				(					snake_case_       : List[str]					,    snake_case_       : int					,    snake_case_       : Dict					,    snake_case_       : int=None					,    snake_case_       : List[Any]=None					,    snake_case_       : Dict=None					,    snake_case_       : Union[str, Any]=None					,    )    ->      int:


     '''simple docstring'''




     if attention_mask is None:
          UpperCAmelCase_           =   tf.cast(tf.math.not_equal(snake_case_					,    config.pad_token_id			)					,    tf.inta			)
     if decoder_attention_mask is None:
          UpperCAmelCase_           =   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_           =   tf.ones((config.encoder_layers, config.encoder_attention_heads)			)
     if decoder_head_mask is None:
          UpperCAmelCase_           =   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        __A      ( UpperCamelCase__		,	UpperCamelCase__		,	unittest.TestCase	):
    a__       :						Tuple          =  (TFLEDForConditionalGeneration, TFLEDModel) if is_tf_available() else ()
    a__       :						Optional[Any]          =  (TFLEDForConditionalGeneration,) if is_tf_available() else ()
    a__       :						Optional[Any]          =  (
        {
            """conversational""": TFLEDForConditionalGeneration,
            """feature-extraction""": TFLEDModel,
            """summarization""": TFLEDForConditionalGeneration,
            """text2text-generation""": TFLEDForConditionalGeneration,
            """translation""": TFLEDForConditionalGeneration,
        }
        if is_tf_available()
        else {}
    )
    a__       :						Optional[Any]          =  True
    a__       :						Optional[Any]          =  False
    a__       :						Tuple          =  False
    a__       :						str          =  False



    def 					_lowercase						(self    :    List[Any] ):
         UpperCAmelCase_           =   TFLEDModelTester(self )
         UpperCAmelCase_           =   ConfigTester(self   ,			config_class=__a )



    def 					_lowercase						(self    :    str ):
         self.config_tester.run_common_tests()



    def 					_lowercase						(self    :    List[Any] ):
         UpperCAmelCase_           =   self.model_tester.prepare_config_and_inputs_for_common()
         self.model_tester.check_decoder_model_past_large_inputs(*__a )



    def 					_lowercase						(self    :    Union[str, Any] ):
         UpperCAmelCase_    , UpperCAmelCase_           =   self.model_tester.prepare_config_and_inputs_for_common()
         UpperCAmelCase_           =   tf.zeros_like(inputs_dict["attention_mask"] )
         UpperCAmelCase_           =   2
         UpperCAmelCase_           =   tf.where(
             tf.range(self.model_tester.seq_length )[None, :] < num_global_attn_indices   ,			1   ,			inputs_dict["global_attention_mask"]   ,			)

         UpperCAmelCase_           =   True
         UpperCAmelCase_           =   self.model_tester.seq_length
         UpperCAmelCase_           =   self.model_tester.encoder_seq_length

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

              if self.is_encoder_decoder:
                   UpperCAmelCase_           =   model_class(__a )
                   UpperCAmelCase_           =   model(self._prepare_for_class(__a   ,			__a ) )
                   self.assertEqual(config.output_hidden_states   ,			__a )
                   check_decoder_attentions_output(__a )

              # Check that output attentions can also be changed via the config
              del inputs_dict["output_attentions"]
              UpperCAmelCase_           =   True
              UpperCAmelCase_           =   model_class(__a )
              UpperCAmelCase_           =   model(self._prepare_for_class(__a   ,			__a ) )
              self.assertEqual(config.output_hidden_states   ,			__a )
              check_encoder_attentions_output(__a )

              # Check attention is always last and order is fine
              UpperCAmelCase_           =   True
              UpperCAmelCase_           =   True
              UpperCAmelCase_           =   model_class(__a )
              UpperCAmelCase_           =   model(self._prepare_for_class(__a   ,			__a ) )

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



    @unittest.skip("LED keeps using potentially symbolic tensors in conditionals and breaks tracing." )
    def 					_lowercase						(self    :    Union[str, Any] ):
         pass






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





def 	lowerCAmelCase_				(					snake_case_       : List[str]			)    ->      Optional[int]:


     '''simple docstring'''




     return tf.constant(snake_case_					,    dtype=tf.intaa			)


SCREAMING_SNAKE_CASE_:     List[str]  =1E-4






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



    def 					_lowercase						(self    :    str ):
         UpperCAmelCase_           =   TFLEDForConditionalGeneration.from_pretrained("allenai/led-base-16384" ).led

         # change to intended input here
         UpperCAmelCase_           =   _long_tensor([512 * [0, 31414, 232, 328, 740, 1140, 12695, 69]] )
         UpperCAmelCase_           =   _long_tensor([128 * [0, 31414, 232, 328, 740, 1140, 12695, 69]] )
         UpperCAmelCase_           =   prepare_led_inputs_dict(model.config   ,			__a   ,			__a )
         UpperCAmelCase_           =   model(**__a )[0]
         UpperCAmelCase_           =   (1, 1024, 768)
         self.assertEqual(output.shape   ,			__a )
         # change to expected output here
         UpperCAmelCase_           =   tf.convert_to_tensor(
             [[2.30_50, 2.82_79, 0.65_31], [-1.84_57, -0.14_55, -3.56_61], [-1.01_86, 0.45_86, -2.20_43]]   ,			)
         tf.debugging.assert_near(output[:, :3, :3]   ,			__a   ,			atol=1E-3 )






    def 					_lowercase						(self    :    List[Any] ):
         UpperCAmelCase_           =   TFLEDForConditionalGeneration.from_pretrained("allenai/led-base-16384" )

         # change to intended input here
         UpperCAmelCase_           =   _long_tensor([512 * [0, 31414, 232, 328, 740, 1140, 12695, 69]] )
         UpperCAmelCase_           =   _long_tensor([128 * [0, 31414, 232, 328, 740, 1140, 12695, 69]] )
         UpperCAmelCase_           =   prepare_led_inputs_dict(model.config   ,			__a   ,			__a )
         UpperCAmelCase_           =   model(**__a )[0]
         UpperCAmelCase_           =   (1, 1024, model.config.vocab_size)
         self.assertEqual(output.shape   ,			__a )
         # change to expected output here
         UpperCAmelCase_           =   tf.convert_to_tensor(
             [[33.65_07, 6.45_72, 16.80_89], [5.87_39, -2.42_38, 11.29_02], [-3.21_39, -4.31_49, 4.27_83]]   ,			)
         tf.debugging.assert_near(output[:, :3, :3]   ,			__a   ,			atol=1E-3   ,			rtol=1E-3 )





'''simple docstring'''





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)
SCREAMING_SNAKE_CASE_:     Optional[int]  =logging.getLogger()

SCREAMING_SNAKE_CASE_:     Union[str, Any]  =logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)




class        __A      ( UpperCamelCase__	):



    def 					_lowercase						(self    :    Optional[Any]   ,			__a    :    str ):
         os.makedirs(__a   ,			exist_ok=__a )
         UpperCAmelCase_           =   {"source": "What is love ?", "target": "life"}
         UpperCAmelCase_           =   {"train": 12, "val": 2, "test": 2}
         for split in ["train", "test", "val"]:
              for field in ["source", "target"]:
                   UpperCAmelCase_           =   "\n".join([contents[field]] * n_lines[split] )
                   with open(os.path.join(__a   ,			f"""{split}.{field}""" )   ,			"w" ) as f:
                        f.write(__a )



    def 					_lowercase						(self    :    Optional[int]   ,			__a    :    int   ,			__a    :    str = "pytorch" ):
         UpperCAmelCase_           =   self.get_auto_remove_tmp_dir()
         UpperCAmelCase_           =   os.path.join(__a   ,			"output" )
         UpperCAmelCase_           =   os.path.join(__a   ,			"data" )
         self._create_dummy_data(data_dir=__a )

         UpperCAmelCase_           =   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_           =   [sys.executable, str(Path(finetune_rag.__file__ ).resolve() )] + testargs
         execute_subprocess_async(__a   ,			env=self.get_env() )

         UpperCAmelCase_           =   os.path.join(__a   ,			"metrics.json" )
         with open(__a ) as f:
              UpperCAmelCase_           =   json.load(__a )
         return result



    @require_torch_gpu
    def 					_lowercase						(self    :    Optional[int] ):
         UpperCAmelCase_           =   self._run_finetune(gpus=1 )
         self.assertGreaterEqual(result["test"][0]["test_avg_em"]   ,			0.2 )



    @require_torch_multi_gpu
    def 					_lowercase						(self    :    Dict ):
         UpperCAmelCase_           =   self._run_finetune(gpus=2 )
         self.assertGreaterEqual(result["test"][0]["test_avg_em"]   ,			0.2 )



    @require_torch_gpu
    @require_ray
    def 					_lowercase						(self    :    Optional[int] ):
         UpperCAmelCase_           =   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 					_lowercase						(self    :    Any ):
         UpperCAmelCase_           =   self._run_finetune(gpus=1   ,			distributed_retriever="ray" )
         self.assertGreaterEqual(result["test"][0]["test_avg_em"]   ,			0.2 )




