





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__    :
      A_						:    Any							=   LEDConfig
      A_						:    Optional[int]							=   {}
      A_						:    Union[str, Any]							=   'gelu'


      def __init__(self	:				Union[str, Any]    ,      a__	:				Union[str, Any]    ,      a__	:				Optional[int]=13    ,      a__	:				Dict=7    ,      a__	:				List[str]=True    ,      a__	:				List[str]=False    ,      a__	:				Union[str, Any]=99    ,      a__	:				int=32    ,      a__	:				Dict=2    ,      a__	:				int=4    ,      a__	:				Optional[int]=37    ,      a__	:				List[Any]=0.1    ,      a__	:				Union[str, Any]=0.1    ,      a__	:				Tuple=20    ,      a__	:				Optional[Any]=2    ,      a__	:				Union[str, Any]=1    ,      a__	:				Optional[Any]=0    ,      a__	:				List[Any]=4    ,      ):





         """simple docstring"""



         __snake_case								=       parent
         __snake_case								=       batch_size
         __snake_case								=       seq_length
         __snake_case								=       is_training
         __snake_case								=       use_labels
         __snake_case								=       vocab_size
         __snake_case								=       hidden_size
         __snake_case								=       num_hidden_layers
         __snake_case								=       num_attention_heads
         __snake_case								=       intermediate_size

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


      def        a			(self	:				Dict				):





         """simple docstring"""



         __snake_case								=       ids_tensor([self.batch_size, self.seq_length - 1]    ,      self.vocab_size				)
         __snake_case								=       tf.expand_dims(tf.constant([self.eos_token_id] * self.batch_size				)    ,      1				)
         __snake_case								=       tf.concat([input_ids, eos_tensor]    ,      axis=1				)

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

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


      def        a			(self	:				Any    ,      a__	:				Dict    ,      a__	:				Optional[Any]				):





         """simple docstring"""



         __snake_case								=       TFLEDModel(config=a__				).get_decoder()
         __snake_case								=       inputs_dict['''input_ids''']

         __snake_case								=       input_ids[:1, :]
         __snake_case								=       inputs_dict['''attention_mask'''][:1, :]
         __snake_case								=       1

         # first forward pass
         __snake_case								=       model(a__    ,      attention_mask=a__    ,      use_cache=a__				)

         __snake_case	,     __snake_case								=       outputs.to_tuple()

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

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

         __snake_case								=       model(a__    ,      attention_mask=a__				)[0]
         __snake_case								=       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
         __snake_case								=       int(ids_tensor((1,)    ,      output_from_past.shape[-1]				)				)
         __snake_case								=       output_from_no_past[:, -3:, random_slice_idx]
         __snake_case								=       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_     :  Any	,				snake_case_     :  int	,				snake_case_     :  Union[str, Any]	,				snake_case_     :  Optional[int]=None	,				snake_case_     :  str=None	,				snake_case_     :  Optional[Any]=None	,				snake_case_     :  Optional[int]=None	,				)				->    List[str]:
   if attention_mask is None:
      __snake_case								=       tf.cast(tf.math.not_equal(snake_case_	,				config.pad_token_id	)	,				tf.inta	)
   if decoder_attention_mask is None:
      __snake_case								=       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:
      __snake_case								=       tf.ones((config.encoder_layers, config.encoder_attention_heads)	)
   if decoder_head_mask is None:
      __snake_case								=       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__    (	_UpperCAmelCase						,				_UpperCAmelCase						,				unittest.TestCase  ):
      A_						:    Tuple							=   (TFLEDForConditionalGeneration, TFLEDModel) if is_tf_available() else ()
      A_						:    int							=   (TFLEDForConditionalGeneration,) if is_tf_available() else ()
      A_						:    List[Any]							=   (
          {
              'conversational': TFLEDForConditionalGeneration,
              'feature-extraction': TFLEDModel,
              'summarization': TFLEDForConditionalGeneration,
              'text2text-generation': TFLEDForConditionalGeneration,
              'translation': TFLEDForConditionalGeneration,
          }
          if is_tf_available()
          else {}
      )
      A_						:    Optional[Any]							=   True
      A_						:    List[Any]							=   False
      A_						:    Optional[Any]							=   False
      A_						:    str							=   False


      def        a			(self	:				Optional[int]				):





         """simple docstring"""



         __snake_case								=       TFLEDModelTester(self				)
         __snake_case								=       ConfigTester(self    ,      config_class=a__				)


      def        a			(self	:				Tuple				):





         """simple docstring"""



         self.config_tester.run_common_tests()


      def        a			(self	:				Any				):





         """simple docstring"""



         __snake_case								=       self.model_tester.prepare_config_and_inputs_for_common()
         self.model_tester.check_decoder_model_past_large_inputs(*a__				)


      def        a			(self	:				Any				):





         """simple docstring"""



         __snake_case	,     __snake_case								=       self.model_tester.prepare_config_and_inputs_for_common()
         __snake_case								=       tf.zeros_like(inputs_dict['''attention_mask''']				)
         __snake_case								=       2
         __snake_case								=       tf.where(
             tf.range(self.model_tester.seq_length				)[None, :] < num_global_attn_indices    ,      1    ,      inputs_dict['''global_attention_mask''']    ,      )

         __snake_case								=       True
         __snake_case								=       self.model_tester.seq_length
         __snake_case								=       self.model_tester.encoder_seq_length

         def check_decoder_attentions_output(a__	:				Dict				):
            __snake_case								=       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[Any]				):
            __snake_case								=       [t.numpy() for t in outputs.encoder_attentions]
            __snake_case								=       [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:
            __snake_case								=       True
            __snake_case								=       False
            __snake_case								=       False
            __snake_case								=       model_class(a__				)
            __snake_case								=       model(self._prepare_for_class(a__    ,      a__				)				)
            __snake_case								=       len(a__				)
            self.assertEqual(config.output_hidden_states    ,      a__				)
            check_encoder_attentions_output(a__				)

            if self.is_encoder_decoder:
               __snake_case								=       model_class(a__				)
               __snake_case								=       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"]
            __snake_case								=       True
            __snake_case								=       model_class(a__				)
            __snake_case								=       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
            __snake_case								=       True
            __snake_case								=       True
            __snake_case								=       model_class(a__				)
            __snake_case								=       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        a			(self	:				List[str]				):





         """simple docstring"""



         pass


      def        a			(self	:				str				):





         """simple docstring"""



         pass






def lowerCamelCase__   (							snake_case_     :  List[Any]	)				->    Optional[Any]:
   return tf.constant(snake_case_	,				dtype=tf.intaa	)


snake_case_           =      1E-4







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


      def        a			(self	:				int				):





         """simple docstring"""



         __snake_case								=       TFLEDForConditionalGeneration.from_pretrained('''allenai/led-base-16384'''				).led

         # change to intended input here
         __snake_case								=       _long_tensor([512 * [0, 3_1414, 232, 328, 740, 1140, 1_2695, 69]]				)
         __snake_case								=       _long_tensor([128 * [0, 3_1414, 232, 328, 740, 1140, 1_2695, 69]]				)
         __snake_case								=       prepare_led_inputs_dict(model.config    ,      a__    ,      a__				)
         __snake_case								=       model(**a__				)[0]
         __snake_case								=       (1, 1024, 768)
         self.assertEqual(output.shape    ,      a__				)
         # change to expected output here
         __snake_case								=       tf.convert_to_tensor(
             [[2.3_0_5_0, 2.8_2_7_9, 0.6_5_3_1], [-1.8_4_5_7, -0.1_4_5_5, -3.5_6_6_1], [-1.0_1_8_6, 0.4_5_8_6, -2.2_0_4_3]]    ,      )
         tf.debugging.assert_near(output[:, :3, :3]    ,      a__    ,      atol=1E-3				)


      def        a			(self	:				List[str]				):





         """simple docstring"""



         __snake_case								=       TFLEDForConditionalGeneration.from_pretrained('''allenai/led-base-16384'''				)

         # change to intended input here
         __snake_case								=       _long_tensor([512 * [0, 3_1414, 232, 328, 740, 1140, 1_2695, 69]]				)
         __snake_case								=       _long_tensor([128 * [0, 3_1414, 232, 328, 740, 1140, 1_2695, 69]]				)
         __snake_case								=       prepare_led_inputs_dict(model.config    ,      a__    ,      a__				)
         __snake_case								=       model(**a__				)[0]
         __snake_case								=       (1, 1024, model.config.vocab_size)
         self.assertEqual(output.shape    ,      a__				)
         # change to expected output here
         __snake_case								=       tf.convert_to_tensor(
             [[3_3.6_5_0_7, 6.4_5_7_2, 1_6.8_0_8_9], [5.8_7_3_9, -2.4_2_3_8, 1_1.2_9_0_2], [-3.2_1_3_9, -4.3_1_4_9, 4.2_7_8_3]]    ,      )
         tf.debugging.assert_near(output[:, :3, :3]    ,      a__    ,      atol=1E-3    ,      rtol=1E-3				)








import argparse
import os
import pickle
import sys

import torch

from transformers import TransfoXLConfig, TransfoXLLMHeadModel, load_tf_weights_in_transfo_xl
from transformers.models.transfo_xl import tokenization_transfo_xl as data_utils
from transformers.models.transfo_xl.tokenization_transfo_xl import CORPUS_NAME, VOCAB_FILES_NAMES
from transformers.utils import CONFIG_NAME, WEIGHTS_NAME, logging


logging.set_verbosity_info()

# We do this to be able to load python 2 datasets pickles
# See e.g. https://stackoverflow.com/questions/2121874/python-pickling-after-changing-a-modules-directory/2121918#2121918
snake_case_           =      data_utils.TransfoXLTokenizer
snake_case_           =      data_utils.TransfoXLCorpus
snake_case_           =      data_utils
snake_case_           =      data_utils






def lowerCamelCase__   (							snake_case_     :  Union[str, Any]	,				snake_case_     :  int	,				snake_case_     :  List[Any]	,				snake_case_     :  int	)				->    Dict:
   if transfo_xl_dataset_file:
      # Convert a pre-processed corpus (see original TensorFlow repo)
      with open(snake_case_	,				'''rb'''	) as fp:
         __snake_case								=       pickle.load(snake_case_	,				encoding='''latin1'''	)
      # Save vocabulary and dataset cache as Dictionaries (should be better than pickles for the long-term)
      __snake_case								=       pytorch_dump_folder_path + '''/''' + VOCAB_FILES_NAMES['''pretrained_vocab_file''']
      print(f"""Save vocabulary to {pytorch_vocab_dump_path}"""	)
      __snake_case								=       corpus.vocab.__dict__
      torch.save(snake_case_	,				snake_case_	)

      __snake_case								=       corpus.__dict__
      corpus_dict_no_vocab.pop('''vocab'''	,				snake_case_	)
      __snake_case								=       pytorch_dump_folder_path + '''/''' + CORPUS_NAME
      print(f"""Save dataset to {pytorch_dataset_dump_path}"""	)
      torch.save(snake_case_	,				snake_case_	)

   if tf_checkpoint_path:
      # Convert a pre-trained TensorFlow model
      __snake_case								=       os.path.abspath(snake_case_	)
      __snake_case								=       os.path.abspath(snake_case_	)

      print(f"""Converting Transformer XL checkpoint from {tf_path} with config at {config_path}."""	)
      # Initialise PyTorch model
      if transfo_xl_config_file == "":
         __snake_case								=       TransfoXLConfig()
      else:
         __snake_case								=       TransfoXLConfig.from_json_file(snake_case_	)
      print(f"""Building PyTorch model from configuration: {config}"""	)
      __snake_case								=       TransfoXLLMHeadModel(snake_case_	)

      __snake_case								=       load_tf_weights_in_transfo_xl(snake_case_	,				snake_case_	,				snake_case_	)
      # Save pytorch-model
      __snake_case								=       os.path.join(snake_case_	,				snake_case_	)
      __snake_case								=       os.path.join(snake_case_	,				snake_case_	)
      print(f"""Save PyTorch model to {os.path.abspath(snake_case_	)}"""	)
      torch.save(model.state_dict()	,				snake_case_	)
      print(f"""Save configuration file to {os.path.abspath(snake_case_	)}"""	)
      with open(snake_case_	,				'''w'''	,				encoding='''utf-8'''	) as f:
         f.write(config.to_json_string()	)


if __name__ == "__main__":
    snake_case_           =      argparse.ArgumentParser()
    parser.add_argument(
        '--pytorch_dump_folder_path',
        default=None,
        type=str,
        required=True,
        help='Path to the folder to store the PyTorch model or dataset/vocab.',
    )
    parser.add_argument(
        '--tf_checkpoint_path',
        default='',
        type=str,
        help='An optional path to a TensorFlow checkpoint path to be converted.',
    )
    parser.add_argument(
        '--transfo_xl_config_file',
        default='',
        type=str,
        help=(
            'An optional config json file corresponding to the pre-trained BERT model. \n'
            'This specifies the model architecture.'
        ),
    )
    parser.add_argument(
        '--transfo_xl_dataset_file',
        default='',
        type=str,
        help='An optional dataset file to be converted in a vocabulary.',
    )
    snake_case_           =      parser.parse_args()
    convert_transfo_xl_checkpoint_to_pytorch(
        args.tf_checkpoint_path,
        args.transfo_xl_config_file,
        args.pytorch_dump_folder_path,
        args.transfo_xl_dataset_file,
    )

