






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







     '''simple docstring'''
     SCREAMING_SNAKE_CASE   						=				LEDConfig
     SCREAMING_SNAKE_CASE   						=				{}
     SCREAMING_SNAKE_CASE   						=				"gelu"




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


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





     def 		lowerCAmelCase_     (				self       :			Optional[Any] ,	UpperCamelCase_       :			str ,	UpperCamelCase_       :			Dict				):


         """simple docstring"""



         __A         =	TFLEDModel(config=UpperCamelCase_				).get_decoder()
         __A         =	inputs_dict["""input_ids"""]

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

         # first forward pass
         __A         =	model(UpperCamelCase_ ,	attention_mask=UpperCamelCase_ ,	use_cache=UpperCamelCase_				)

         __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(UpperCamelCase_ ,	attention_mask=UpperCamelCase_				)[0]
         __A         =	model(UpperCamelCase_ ,	attention_mask=UpperCamelCase_ ,	past_key_values=UpperCamelCase_				)[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(UpperCamelCase_ ,	UpperCamelCase_ ,	rtol=1e-3				)



def 			_SCREAMING_SNAKE_CASE						( __lowercase   :			Optional[Any]			,				__lowercase   :			str			,				__lowercase   :			Optional[int]			,				__lowercase   :			Union[str, Any]=None			,				__lowercase   :			str=None			,				__lowercase   :			List[Any]=None			,				__lowercase   :			Union[str, Any]=None			,				)						->      Optional[int]:





    """simple docstring"""

    if attention_mask is None:
        __A         =	tf.cast(tf.math.not_equal(__lowercase			,				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       __lowercase      (	lowercase_			,      lowercase_			,      unittest.TestCase  ):







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




     def 		lowerCAmelCase_     (				self       :			Tuple				):


         """simple docstring"""



         __A         =	TFLEDModelTester(self				)
         __A         =	ConfigTester(self ,	config_class=UpperCamelCase_				)




     def 		lowerCAmelCase_     (				self       :			str				):


         """simple docstring"""



         self.config_tester.run_common_tests()




     def 		lowerCAmelCase_     (				self       :			List[str]				):


         """simple docstring"""



         __A         =	self.model_tester.prepare_config_and_inputs_for_common()
         self.model_tester.check_decoder_model_past_large_inputs(*UpperCamelCase_				)




     def 		lowerCAmelCase_     (				self       :			Dict				):


         """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(UpperCamelCase_       :			str				):
             __A         =	outputs.decoder_attentions
             self.assertEqual(len(UpperCamelCase_				) ,	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(UpperCamelCase_       :			Any				):
             __A         =	[t.numpy() for t in outputs.encoder_attentions]
             __A         =	[t.numpy() for t in outputs.encoder_global_attentions]
             self.assertEqual(len(UpperCamelCase_				) ,	self.model_tester.num_hidden_layers				)
             self.assertEqual(len(UpperCamelCase_				) ,	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(UpperCamelCase_				)
             __A         =	model(self._prepare_for_class(UpperCamelCase_ ,	UpperCamelCase_				)				)
             __A         =	len(UpperCamelCase_				)
             self.assertEqual(config.output_hidden_states ,	UpperCamelCase_				)
             check_encoder_attentions_output(UpperCamelCase_				)

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

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

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

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




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


         """simple docstring"""



         pass





     def 		lowerCAmelCase_     (				self       :			List[str]				):


         """simple docstring"""



         pass



def 			_SCREAMING_SNAKE_CASE						( __lowercase   :			int    )						->      List[str]:





    """simple docstring"""

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


__a      :       List[str]				  =   1e-4






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







     '''simple docstring'''




     def 		lowerCAmelCase_     (				self       :			Tuple				):


         """simple docstring"""



         __A         =	TFLEDForConditionalGeneration.from_pretrained("""allenai/led-base-16384"""				).led

         # change to intended input here
         __A         =	_long_tensor([512 * [0, 31_414, 232, 328, 740, 1_140, 12_695, 69]]				)
         __A         =	_long_tensor([128 * [0, 31_414, 232, 328, 740, 1_140, 12_695, 69]]				)
         __A         =	prepare_led_inputs_dict(model.config ,	UpperCamelCase_ ,	UpperCamelCase_				)
         __A         =	model(**UpperCamelCase_				)[0]
         __A         =	(1, 1_024, 768)
         self.assertEqual(output.shape ,	UpperCamelCase_				)
         # change to expected output here
         __A         =	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] ,	UpperCamelCase_ ,	atol=1e-3				)





     def 		lowerCAmelCase_     (				self       :			Union[str, Any]				):


         """simple docstring"""



         __A         =	TFLEDForConditionalGeneration.from_pretrained("""allenai/led-base-16384"""				)

         # change to intended input here
         __A         =	_long_tensor([512 * [0, 31_414, 232, 328, 740, 1_140, 12_695, 69]]				)
         __A         =	_long_tensor([128 * [0, 31_414, 232, 328, 740, 1_140, 12_695, 69]]				)
         __A         =	prepare_led_inputs_dict(model.config ,	UpperCamelCase_ ,	UpperCamelCase_				)
         __A         =	model(**UpperCamelCase_				)[0]
         __A         =	(1, 1_024, model.config.vocab_size)
         self.assertEqual(output.shape ,	UpperCamelCase_				)
         # change to expected output here
         __A         =	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] ,	UpperCamelCase_ ,	atol=1e-3 ,	rtol=1e-3				)












from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, Encoder, VectorQuantizer


@dataclass
class       __lowercase      (	lowercase_  ):







     '''simple docstring'''
     SCREAMING_SNAKE_CASE   						=				42


class       __lowercase      (	lowercase_			,      lowercase_  ):







     '''simple docstring'''




     @register_to_config
     def __init__(				self       :			List[Any] ,	UpperCamelCase_       :			int = 3 ,	UpperCamelCase_       :			int = 3 ,	UpperCamelCase_       :			Tuple[str] = ("DownEncoderBlock2D",) ,	UpperCamelCase_       :			Tuple[str] = ("UpDecoderBlock2D",) ,	UpperCamelCase_       :			Tuple[int] = (64,) ,	UpperCamelCase_       :			int = 1 ,	UpperCamelCase_       :			str = "silu" ,	UpperCamelCase_       :			int = 3 ,	UpperCamelCase_       :			int = 32 ,	UpperCamelCase_       :			int = 256 ,	UpperCamelCase_       :			int = 32 ,	UpperCamelCase_       :			Optional[int] = None ,	UpperCamelCase_       :			float = 0.18215 ,	UpperCamelCase_       :			str = "group" ,	):


         """simple docstring"""



         super().__init__()

         # pass init params to Encoder
         __A         =	Encoder(
             in_channels=UpperCamelCase_ ,	out_channels=UpperCamelCase_ ,	down_block_types=UpperCamelCase_ ,	block_out_channels=UpperCamelCase_ ,	layers_per_block=UpperCamelCase_ ,	act_fn=UpperCamelCase_ ,	norm_num_groups=UpperCamelCase_ ,	double_z=UpperCamelCase_ ,	)

         __A         =	vq_embed_dim if vq_embed_dim is not None else latent_channels

         __A         =	nn.Convad(UpperCamelCase_ ,	UpperCamelCase_ ,	1				)
         __A         =	VectorQuantizer(UpperCamelCase_ ,	UpperCamelCase_ ,	beta=0.25 ,	remap=UpperCamelCase_ ,	sane_index_shape=UpperCamelCase_				)
         __A         =	nn.Convad(UpperCamelCase_ ,	UpperCamelCase_ ,	1				)

         # pass init params to Decoder
         __A         =	Decoder(
             in_channels=UpperCamelCase_ ,	out_channels=UpperCamelCase_ ,	up_block_types=UpperCamelCase_ ,	block_out_channels=UpperCamelCase_ ,	layers_per_block=UpperCamelCase_ ,	act_fn=UpperCamelCase_ ,	norm_num_groups=UpperCamelCase_ ,	norm_type=UpperCamelCase_ ,	)




     @apply_forward_hook
     def 		lowerCAmelCase_     (				self       :			List[str] ,	UpperCamelCase_       :			torch.FloatTensor ,	UpperCamelCase_       :			bool = True				):


         """simple docstring"""



         __A         =	self.encoder(UpperCamelCase_				)
         __A         =	self.quant_conv(UpperCamelCase_				)

         if not return_dict:
             return (h,)

         return VQEncoderOutput(latents=UpperCamelCase_				)




     @apply_forward_hook
     def 		lowerCAmelCase_     (				self       :			List[str] ,	UpperCamelCase_       :			torch.FloatTensor ,	UpperCamelCase_       :			bool = False ,	UpperCamelCase_       :			bool = True				):


         """simple docstring"""



         if not force_not_quantize:
             __A    ,	__A    ,	__A         =	self.quantize(UpperCamelCase_				)
         else:
             __A         =	h
         __A         =	self.post_quant_conv(UpperCamelCase_				)
         __A         =	self.decoder(UpperCamelCase_ ,	quant if self.config.norm_type == """spatial""" else None				)

         if not return_dict:
             return (dec,)

         return DecoderOutput(sample=UpperCamelCase_				)





     def 		lowerCAmelCase_     (				self       :			Any ,	UpperCamelCase_       :			torch.FloatTensor ,	UpperCamelCase_       :			bool = True				):


         """simple docstring"""



         __A         =	sample
         __A         =	self.encode(UpperCamelCase_				).latents
         __A         =	self.decode(UpperCamelCase_				).sample

         if not return_dict:
             return (dec,)

         return DecoderOutput(sample=UpperCamelCase_				)




