



from __future__ import annotations

import unittest
import warnings

from transformers import AutoTokenizer, MarianConfig, MarianTokenizer, TranslationPipeline, is_tf_available
from transformers.testing_utils import require_sentencepiece, require_tf, require_tokenizers, slow
from transformers.utils import cached_property

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 TFAutoModelForSeqaSeqLM, TFMarianModel, TFMarianMTModel






@require_tf
class 	_a			:
			_UpperCamelCase:   List[Any]	     =		MarianConfig
			_UpperCamelCase:   List[Any]	     =		{}
			_UpperCamelCase:   Any	     =		"gelu"






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

									lowerCAmelCase	:   int					     =					hidden_dropout_prob
									lowerCAmelCase	:   Tuple					     =					attention_probs_dropout_prob
									lowerCAmelCase	:   Union[str, Any]					     =					max_position_embeddings
									lowerCAmelCase	:   Optional[int]					     =					eos_token_id
									lowerCAmelCase	:   Optional[int]					     =					pad_token_id
									lowerCAmelCase	:   Any					     =					bos_token_id






			def 							_snake_case   (    self )	->					List[Any]:
									lowerCAmelCase	:   List[Any]					     =					ids_tensor([self.batch_size, self.seq_length - 1]    ,  self.vocab_size )
									lowerCAmelCase	:   Tuple					     =					tf.expand_dims(tf.constant([self.eos_token_id] * self.batch_size )    ,  1 )
									lowerCAmelCase	:   List[Any]					     =					tf.concat([input_ids, eos_tensor]    ,  axis=1 )

									lowerCAmelCase	:   int					     =					ids_tensor([self.batch_size, self.seq_length]    ,  self.vocab_size )

									lowerCAmelCase	:   Tuple					     =					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    ,  **self.config_updates    ,  )
									lowerCAmelCase	:   List[str]					     =					prepare_marian_inputs_dict(lowercase_    ,  lowercase_    ,  lowercase_ )
									return config, inputs_dict






			def 							_snake_case   (    self    ,  lowercase_    ,  lowercase_ )	->					Any:
									lowerCAmelCase	:   List[str]					     =					TFMarianModel(config=lowercase_ ).get_decoder()
									lowerCAmelCase	:   Any					     =					inputs_dict["""input_ids"""]

									lowerCAmelCase	:   Optional[Any]					     =					input_ids[:1, :]
									lowerCAmelCase	:   Union[str, Any]					     =					inputs_dict["""attention_mask"""][:1, :]
									lowerCAmelCase	:   List[str]					     =					inputs_dict["""head_mask"""]
									lowerCAmelCase	:   List[Any]					     =					1

									# first forward pass
									lowerCAmelCase	:   Optional[Any]					     =					model(lowercase_    ,  attention_mask=lowercase_    ,  head_mask=lowercase_    ,  use_cache=lowercase_ )

									lowerCAmelCase     ,	lowerCAmelCase	:   List[Any]					     =					outputs.to_tuple()

									# create hypothetical next token and extent to next_input_ids
									lowerCAmelCase	:   List[Any]					     =					ids_tensor((self.batch_size, 3)    ,  config.vocab_size )
									lowerCAmelCase	:   Optional[Any]					     =					tf.cast(ids_tensor((self.batch_size, 3)    ,  2 )    ,  tf.inta )

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

									lowerCAmelCase	:   Optional[int]					     =					model(lowercase_    ,  attention_mask=lowercase_ )[0]
									lowerCAmelCase	:   Optional[Any]					     =					model(lowercase_    ,  attention_mask=lowercase_    ,  past_key_values=lowercase_ )[0]

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

									# select random slice
									lowerCAmelCase	:   Tuple					     =					int(ids_tensor((1,)    ,  output_from_past.shape[-1] ) )
									lowerCAmelCase	:   Optional[Any]					     =					output_from_no_past[:, -3:, random_slice_idx]
									lowerCAmelCase	:   List[Any]					     =					output_from_past[:, :, random_slice_idx]

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







def  _UpperCAmelCase  (				SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__			,SCREAMING_SNAKE_CASE__=None			,SCREAMING_SNAKE_CASE__=None			,SCREAMING_SNAKE_CASE__=None			,SCREAMING_SNAKE_CASE__=None			,SCREAMING_SNAKE_CASE__=None			,):


						'''simple docstring'''




						if attention_mask is None:
												lowerCAmelCase	:   Tuple					     =					tf.cast(tf.math.not_equal(SCREAMING_SNAKE_CASE__			,config.pad_token_id  )			,tf.inta  )
						if decoder_attention_mask is None:
												lowerCAmelCase	:   List[str]					     =					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:
												lowerCAmelCase	:   Union[str, Any]					     =					tf.ones((config.encoder_layers, config.encoder_attention_heads)  )
						if decoder_head_mask is None:
												lowerCAmelCase	:   Dict					     =					tf.ones((config.decoder_layers, config.decoder_attention_heads)  )
						if cross_attn_head_mask is None:
												lowerCAmelCase	:   int					     =					tf.ones((config.decoder_layers, config.decoder_attention_heads)  )
						return {
						    "input_ids": input_ids,
						    "decoder_input_ids": decoder_input_ids,
						    "attention_mask": attention_mask,
						    "decoder_attention_mask": decoder_attention_mask,
						    "head_mask": head_mask,
						    "decoder_head_mask": decoder_head_mask,
						    "cross_attn_head_mask": cross_attn_head_mask,
						}






@require_tf
class 	_a			(      snake_case_     , snake_case_     , unittest.TestCase  ):
			_UpperCamelCase:   Tuple	     =		(TFMarianMTModel, TFMarianModel) if is_tf_available() else ()
			_UpperCamelCase:   Dict	     =		(TFMarianMTModel,) if is_tf_available() else ()
			_UpperCamelCase:   Optional[int]	     =		(
			    {
			        "conversational": TFMarianMTModel,
			        "feature-extraction": TFMarianModel,
			        "summarization": TFMarianMTModel,
			        "text2text-generation": TFMarianMTModel,
			        "translation": TFMarianMTModel,
			    }
			    if is_tf_available()
			    else {}
			)
			_UpperCamelCase:   int	     =		True
			_UpperCamelCase:   Any	     =		False
			_UpperCamelCase:   Dict	     =		False






			def 							_snake_case   (    self )	->					Any:
									lowerCAmelCase	:   Any					     =					TFMarianModelTester(self )
									lowerCAmelCase	:   str					     =					ConfigTester(self    ,  config_class=lowercase_ )






			def 							_snake_case   (    self )	->					Any:
									self.config_tester.run_common_tests()






			def 							_snake_case   (    self )	->					Any:
									lowerCAmelCase	:   Optional[Any]					     =					self.model_tester.prepare_config_and_inputs_for_common()
									self.model_tester.check_decoder_model_past_large_inputs(*lowercase_ )






@require_tf
class 	_a			(      unittest.TestCase  ):
			_UpperCamelCase:   str	     =		1000  # show more chars for failing integration tests






			@classmethod
			def 							_snake_case   (    cls )	->					None:
									lowerCAmelCase	:   Optional[int]					     =					f"""Helsinki-NLP/opus-mt-{cls.src}-{cls.tgt}"""
									return cls






			@cached_property
			def 							_snake_case   (    self )	->					MarianTokenizer:
									return AutoTokenizer.from_pretrained(self.model_name )






			@property
			def 							_snake_case   (    self )	->					int:
									return self.tokenizer.eos_token_id






			@cached_property
			def 							_snake_case   (    self )	->					Optional[int]:
									warnings.simplefilter("""error""" )
									lowerCAmelCase	:   TFMarianMTModel					     =					TFAutoModelForSeqaSeqLM.from_pretrained(self.model_name )
									assert isinstance(lowercase_    ,  lowercase_ )
									lowerCAmelCase	:   Dict					     =					model.config
									self.assertListEqual(c.bad_words_ids    ,  [[c.pad_token_id]] )
									self.assertEqual(c.max_length    ,  512 )
									self.assertEqual(c.decoder_start_token_id    ,  c.pad_token_id )
									return model






			def 							_snake_case   (    self    ,  **lowercase_ )	->					List[str]:
									lowerCAmelCase	:   Dict					     =					self.translate_src_text(**lowercase_ )
									self.assertListEqual(self.expected_text    ,  lowercase_ )






			def 							_snake_case   (    self    ,  **lowercase_ )	->					Dict:
									lowerCAmelCase	:   str					     =					self.tokenizer(self.src_text    ,  **lowercase_    ,  padding=lowercase_    ,  return_tensors="""tf""" )
									lowerCAmelCase	:   str					     =					self.model.generate(
									    model_inputs.input_ids    ,  attention_mask=model_inputs.attention_mask    ,  num_beams=2    ,  max_length=128 )
									lowerCAmelCase	:   Union[str, Any]					     =					self.tokenizer.batch_decode(generated_ids.numpy()    ,  skip_special_tokens=lowercase_ )
									return generated_words






@require_sentencepiece
@require_tokenizers
@require_tf
class 	_a			(      snake_case_  ):
			_UpperCamelCase:   Dict	     =		"mt"
			_UpperCamelCase:   Any	     =		"en"
			_UpperCamelCase:   Optional[Any]	     =		["Billi messu b'mod ġentili, Ġesù fejjaq raġel li kien milqut bil - marda kerha tal - ġdiem."]
			_UpperCamelCase:   int	     =		["Touching gently, Jesus healed a man who was affected by the sad disease of leprosy."]






			@unittest.skip("""Skipping until #12647 is resolved.""" )
			@slow
			def 							_snake_case   (    self )	->					List[str]:
									self._assert_generated_batch_equal_expected()






@require_sentencepiece
@require_tokenizers
@require_tf
class 	_a			(      snake_case_  ):
			_UpperCamelCase:   Dict	     =		"en"
			_UpperCamelCase:   Any	     =		"zh"
			_UpperCamelCase:   Any	     =		["My name is Wolfgang and I live in Berlin"]
			_UpperCamelCase:   Optional[Any]	     =		["我叫沃尔夫冈 我住在柏林"]






			@unittest.skip("""Skipping until #12647 is resolved.""" )
			@slow
			def 							_snake_case   (    self )	->					str:
									self._assert_generated_batch_equal_expected()






@require_sentencepiece
@require_tokenizers
@require_tf
class 	_a			(      snake_case_  ):
			_UpperCamelCase:   Optional[Any]	     =		"en"
			_UpperCamelCase:   Union[str, Any]	     =		"ROMANCE"
			_UpperCamelCase:   List[Any]	     =		[
			    ">>fr<< Don't spend so much time watching TV.",
			    ">>pt<< Your message has been sent.",
			    ">>es<< He's two years older than me.",
			]
			_UpperCamelCase:   Dict	     =		[
			    "Ne passez pas autant de temps à regarder la télé.",
			    "A sua mensagem foi enviada.",
			    "Es dos años más viejo que yo.",
			]






			@unittest.skip("""Skipping until #12647 is resolved.""" )
			@slow
			def 							_snake_case   (    self )	->					Any:
									self._assert_generated_batch_equal_expected()






			@unittest.skip("""Skipping until #12647 is resolved.""" )
			@slow
			def 							_snake_case   (    self )	->					Optional[Any]:
									lowerCAmelCase	:   List[Any]					     =					TranslationPipeline(self.model    ,  self.tokenizer    ,  framework="""tf""" )
									lowerCAmelCase	:   List[str]					     =					pipeline(self.src_text )
									self.assertEqual(self.expected_text    ,  [x["""translation_text"""] for x in output] )






from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_sentencepiece_available,
    is_speech_available,
    is_torch_available,
)


lowerCAmelCase					:				Optional[int]   	={
    'configuration_speech_to_text_2': ['SPEECH_TO_TEXT_2_PRETRAINED_CONFIG_ARCHIVE_MAP', 'Speech2Text2Config'],
    'processing_speech_to_text_2': ['Speech2Text2Processor'],
    'tokenization_speech_to_text_2': ['Speech2Text2Tokenizer'],
}


try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		lowerCAmelCase					:				int   	=[
		    'SPEECH_TO_TEXT_2_PRETRAINED_MODEL_ARCHIVE_LIST',
		    'Speech2Text2ForCausalLM',
		    'Speech2Text2PreTrainedModel',
		]


if TYPE_CHECKING:
		from .configuration_speech_to_text_a import SPEECH_TO_TEXT_2_PRETRAINED_CONFIG_ARCHIVE_MAP, SpeechaTextaConfig
		from .processing_speech_to_text_a import SpeechaTextaProcessor
		from .tokenization_speech_to_text_a import SpeechaTextaTokenizer

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_speech_to_text_a import (
				    SPEECH_TO_TEXT_2_PRETRAINED_MODEL_ARCHIVE_LIST,
				    SpeechaTextaForCausalLM,
				    SpeechaTextaPreTrainedModel,
				)

else:
		import sys

		lowerCAmelCase					:				Union[str, Any]   	=_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)

