




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					:
							lowercase_           =       LEDConfig
							lowercase_           =       {}
							lowercase_           =       '''gelu'''







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


														"""simple docstring"""


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

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







							def      __lowerCAmelCase    (						self :						Any      )    ->							Tuple:


														"""simple docstring"""


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

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

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







							def      __lowerCAmelCase    (						self :						str							,    lowerCAmelCase_ :						str							,    lowerCAmelCase_ :						str      )    ->							Dict:


														"""simple docstring"""


														_a			  =       TFLEDModel(config=snake_case__      ).get_decoder()
														_a			  =       inputs_dict['''input_ids''']

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

														# first forward pass
														_a			  =       model(snake_case__							,    attention_mask=snake_case__							,    use_cache=snake_case__      )

														_a  ,							_a			  =       outputs.to_tuple()

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

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

														_a			  =       model(snake_case__							,    attention_mask=snake_case__      )[0]
														_a			  =       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
														_a			  =       int(ids_tensor((1,)							,    output_from_past.shape[-1]      )      )
														_a			  =       output_from_no_past[:, -3:, random_slice_idx]
														_a			  =       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 	snake_case_					(UpperCamelCase :						Dict				,		UpperCamelCase :						Optional[Any]				,		UpperCamelCase :						Optional[int]				,		UpperCamelCase :						Any=None				,		UpperCamelCase :						List[str]=None				,		UpperCamelCase :						Tuple=None				,		UpperCamelCase :						str=None				,		):





							'''simple docstring'''


							if attention_mask is None:
														_a			  =       tf.cast(tf.math.not_equal(lowerCAmelCase__				,		config.pad_token_id   )				,		tf.inta   )
							if decoder_attention_mask is None:
														_a			  =       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:
														_a			  =       tf.ones((config.encoder_layers, config.encoder_attention_heads)   )
							if decoder_head_mask is None:
														_a			  =       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					(	__a  ,__a  ,unittest.TestCase					):
							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 :						Union[str, Any]      )    ->							Optional[int]:


														"""simple docstring"""


														_a			  =       TFLEDModelTester(self      )
														_a			  =       ConfigTester(self							,    config_class=snake_case__      )







							def      __lowerCAmelCase    (						self :						Tuple      )    ->							List[Any]:


														"""simple docstring"""


														self.config_tester.run_common_tests()







							def      __lowerCAmelCase    (						self :						Any      )    ->							List[Any]:


														"""simple docstring"""


														_a			  =       self.model_tester.prepare_config_and_inputs_for_common()
														self.model_tester.check_decoder_model_past_large_inputs(*snake_case__      )







							def      __lowerCAmelCase    (						self :						Any      )    ->							Optional[int]:


														"""simple docstring"""


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

														_a			  =       True
														_a			  =       self.model_tester.seq_length
														_a			  =       self.model_tester.encoder_seq_length

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

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


														"""simple docstring"""


														pass







							def      __lowerCAmelCase    (						self :						Dict      )    ->							List[Any]:


														"""simple docstring"""


														pass



def 	snake_case_					(UpperCamelCase :						List[Any]   ):





							'''simple docstring'''


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


_snake_case   :   Union[str, Any]     							=	1E-4

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







							def      __lowerCAmelCase    (						self :						Optional[int]      )    ->							int:


														"""simple docstring"""


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

														# change to intended input here
														_a			  =       _long_tensor([5_12 * [0, 3_14_14, 2_32, 3_28, 7_40, 11_40, 1_26_95, 69]]      )
														_a			  =       _long_tensor([1_28 * [0, 3_14_14, 2_32, 3_28, 7_40, 11_40, 1_26_95, 69]]      )
														_a			  =       prepare_led_inputs_dict(model.config							,    snake_case__							,    snake_case__      )
														_a			  =       model(**snake_case__      )[0]
														_a			  =       (1, 10_24, 7_68)
														self.assertEqual(output.shape							,    snake_case__      )
														# change to expected output here
														_a			  =       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]							,    snake_case__							,    atol=1e-3      )







							def      __lowerCAmelCase    (						self :						List[str]      )    ->							List[Any]:


														"""simple docstring"""


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

														# change to intended input here
														_a			  =       _long_tensor([5_12 * [0, 3_14_14, 2_32, 3_28, 7_40, 11_40, 1_26_95, 69]]      )
														_a			  =       _long_tensor([1_28 * [0, 3_14_14, 2_32, 3_28, 7_40, 11_40, 1_26_95, 69]]      )
														_a			  =       prepare_led_inputs_dict(model.config							,    snake_case__							,    snake_case__      )
														_a			  =       model(**snake_case__      )[0]
														_a			  =       (1, 10_24, model.config.vocab_size)
														self.assertEqual(output.shape							,    snake_case__      )
														# change to expected output here
														_a			  =       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]							,    snake_case__							,    atol=1e-3							,    rtol=1e-3      )







'''simple docstring'''


import argparse
import os

import numpy as np
import tensorflow as tf
import torch

from transformers import BertModel
def 	snake_case_					(UpperCamelCase :						BertModel				,		UpperCamelCase :						str				,		UpperCamelCase :						str   ):





							'''simple docstring'''


							_a			  =       ('''dense.weight''', '''attention.self.query''', '''attention.self.key''', '''attention.self.value''')

							_a			  =       (
							    ('''layer.''', '''layer_'''),
							    ('''word_embeddings.weight''', '''word_embeddings'''),
							    ('''position_embeddings.weight''', '''position_embeddings'''),
							    ('''token_type_embeddings.weight''', '''token_type_embeddings'''),
							    ('''.''', '''/'''),
							    ('''LayerNorm/weight''', '''LayerNorm/gamma'''),
							    ('''LayerNorm/bias''', '''LayerNorm/beta'''),
							    ('''weight''', '''kernel'''),
							)

							if not os.path.isdir(UpperCamelCase   ):
														os.makedirs(UpperCamelCase   )

							_a			  =       model.state_dict()

							def to_tf_var_name(UpperCamelCase :						str   ):
														for patt, repl in iter(UpperCamelCase   ):
																					_a			  =       name.replace(UpperCamelCase				,		UpperCamelCase   )
														return f'bert/{name}'

							def create_tf_var(UpperCamelCase :						np.ndarray				,		UpperCamelCase :						str				,		UpperCamelCase :						tf.Session   ):
														_a			  =       tf.dtypes.as_dtype(tensor.dtype   )
														_a			  =       tf.get_variable(dtype=UpperCamelCase				,		shape=tensor.shape				,		name=UpperCamelCase				,		initializer=tf.zeros_initializer()   )
														session.run(tf.variables_initializer([tf_var]   )   )
														session.run(UpperCamelCase   )
														return tf_var

							tf.reset_default_graph()
							with tf.Session() as session:
														for var_name in state_dict:
																					_a			  =       to_tf_var_name(UpperCamelCase   )
																					_a			  =       state_dict[var_name].numpy()
																					if any(x in var_name for x in tensors_to_transpose   ):
																												_a			  =       torch_tensor.T
																					_a			  =       create_tf_var(tensor=UpperCamelCase				,		name=UpperCamelCase				,		session=UpperCamelCase   )
																					tf.keras.backend.set_value(UpperCamelCase				,		UpperCamelCase   )
																					_a			  =       session.run(UpperCamelCase   )
																					print(f'Successfully created {tf_name}: {np.allclose(UpperCamelCase				,		UpperCamelCase   )}'   )

														_a			  =       tf.train.Saver(tf.trainable_variables()   )
														saver.save(UpperCamelCase				,		os.path.join(UpperCamelCase				,		model_name.replace('''-'''				,		'''_'''   ) + '''.ckpt'''   )   )
def 	snake_case_					(UpperCamelCase :						Tuple=None   ):





							'''simple docstring'''


							_a			  =       argparse.ArgumentParser()
							parser.add_argument('''--model_name'''				,		type=UpperCamelCase				,		required=UpperCamelCase				,		help='''model name e.g. bert-base-uncased'''   )
							parser.add_argument(
							    '''--cache_dir'''				,		type=UpperCamelCase				,		default=UpperCamelCase				,		required=UpperCamelCase				,		help='''Directory containing pytorch model'''   )
							parser.add_argument('''--pytorch_model_path'''				,		type=UpperCamelCase				,		required=UpperCamelCase				,		help='''/path/to/<pytorch-model-name>.bin'''   )
							parser.add_argument('''--tf_cache_dir'''				,		type=UpperCamelCase				,		required=UpperCamelCase				,		help='''Directory in which to save tensorflow model'''   )
							_a			  =       parser.parse_args(UpperCamelCase   )

							_a			  =       BertModel.from_pretrained(
							    pretrained_model_name_or_path=args.model_name				,		state_dict=torch.load(args.pytorch_model_path   )				,		cache_dir=args.cache_dir				,		)

							convert_pytorch_checkpoint_to_tf(model=UpperCamelCase				,		ckpt_dir=args.tf_cache_dir				,		model_name=args.model_name   )


if __name__ == "__main__":
				main()

