


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







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__ :					Any   =							MarianConfig
						SCREAMING_SNAKE_CASE__ :					Optional[int]   =							{}
						SCREAMING_SNAKE_CASE__ :					Optional[Any]   =							'gelu'




						def __init__(			self				,  __lowercase				,  __lowercase=1_3				,  __lowercase=7				,  __lowercase=True				,  __lowercase=False				,  __lowercase=9_9				,  __lowercase=3_2				,  __lowercase=2				,  __lowercase=4				,  __lowercase=3_7				,  __lowercase=0.1				,  __lowercase=0.1				,  __lowercase=2_0				,  __lowercase=2				,  __lowercase=1				,  __lowercase=0				,  )   ->      Optional[int]:
										lowerCAmelCase_	: List[str]      			=			parent
										lowerCAmelCase_	: Tuple      			=			batch_size
										lowerCAmelCase_	: Any      			=			seq_length
										lowerCAmelCase_	: Optional[Any]      			=			is_training
										lowerCAmelCase_	: Union[str, Any]      			=			use_labels
										lowerCAmelCase_	: Union[str, Any]      			=			vocab_size
										lowerCAmelCase_	: int      			=			hidden_size
										lowerCAmelCase_	: List[Any]      			=			num_hidden_layers
										lowerCAmelCase_	: Union[str, Any]      			=			num_attention_heads
										lowerCAmelCase_	: Optional[Any]      			=			intermediate_size

										lowerCAmelCase_	: List[Any]      			=			hidden_dropout_prob
										lowerCAmelCase_	: List[str]      			=			attention_probs_dropout_prob
										lowerCAmelCase_	: str      			=			max_position_embeddings
										lowerCAmelCase_	: Optional[int]      			=			eos_token_id
										lowerCAmelCase_	: Dict      			=			pad_token_id
										lowerCAmelCase_	: Any      			=			bos_token_id




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

										lowerCAmelCase_	: List[Any]      			=			ids_tensor([self.batch_size, self.seq_length]				,  self.vocab_size							)

										lowerCAmelCase_	: int      			=			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_	: Dict      			=			prepare_marian_inputs_dict(__lowercase				,  __lowercase				,  __lowercase							)
										return config, inputs_dict




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

										lowerCAmelCase_	: Any      			=			input_ids[:1, :]
										lowerCAmelCase_	: int      			=			inputs_dict["attention_mask"][:1, :]
										lowerCAmelCase_	: List[Any]      			=			inputs_dict["head_mask"]
										lowerCAmelCase_	: List[str]      			=			1

										# first forward pass
										lowerCAmelCase_	: List[str]      			=			model(__lowercase				,  attention_mask=__lowercase				,  head_mask=__lowercase				,  use_cache=__lowercase							)

										lowerCAmelCase_	: Tuple      			=			outputs.to_tuple()

										# create hypothetical next token and extent to next_input_ids
										lowerCAmelCase_	: Union[str, 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_	: Any      			=			tf.concat([input_ids, next_tokens]				,  axis=-1							)
										lowerCAmelCase_	: Dict      			=			tf.concat([attention_mask, next_attn_mask]				,  axis=-1							)

										lowerCAmelCase_	: List[str]      			=			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_	: Union[str, Any]      			=			int(ids_tensor((1,)				,  output_from_past.shape[-1]							)							)
										lowerCAmelCase_	: Tuple      			=			output_from_no_past[:, -3:, random_slice_idx]
										lowerCAmelCase_	: Optional[Any]      			=			output_from_past[:, :, random_slice_idx]

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






def      lowerCAmelCase	(						lowerCAmelCase_	,				lowerCAmelCase_	,				lowerCAmelCase_	,				lowerCAmelCase_=None	,				lowerCAmelCase_=None	,				lowerCAmelCase_=None	,				lowerCAmelCase_=None	,				lowerCAmelCase_=None	,				)->	Tuple:




				if attention_mask is None:
								lowerCAmelCase_	: Optional[Any]      			=			tf.cast(tf.math.not_equal(_UpperCamelCase	,				config.pad_token_id	)	,				tf.inta	)
				if decoder_attention_mask is None:
								lowerCAmelCase_	: Optional[int]      			=			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_	: Tuple      			=			tf.ones((config.encoder_layers, config.encoder_attention_heads)	)
				if decoder_head_mask is None:
								lowerCAmelCase_	: int      			=			tf.ones((config.decoder_layers, config.decoder_attention_heads)	)
				if cross_attn_head_mask is None:
								lowerCAmelCase_	: Optional[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 			snake_case__(							UpperCAmelCase__,  UpperCAmelCase__,  unittest.TestCase							):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__ :					Any   =							(TFMarianMTModel, TFMarianModel) if is_tf_available() else ()
						SCREAMING_SNAKE_CASE__ :					Dict   =							(TFMarianMTModel,) if is_tf_available() else ()
						SCREAMING_SNAKE_CASE__ :					List[Any]   =							(
						    {
						        'conversational': TFMarianMTModel,
						        'feature-extraction': TFMarianModel,
						        'summarization': TFMarianMTModel,
						        'text2text-generation': TFMarianMTModel,
						        'translation': TFMarianMTModel,
						    }
						    if is_tf_available()
						    else {}
						)
						SCREAMING_SNAKE_CASE__ :					List[str]   =							True
						SCREAMING_SNAKE_CASE__ :					List[str]   =							False
						SCREAMING_SNAKE_CASE__ :					str   =							False




						def   lowercase_	(			self							)   ->      Optional[Any]:
										lowerCAmelCase_	: Any      			=			TFMarianModelTester(self							)
										lowerCAmelCase_	: Union[str, Any]      			=			ConfigTester(self				,  config_class=__lowercase							)




						def   lowercase_	(			self							)   ->      List[str]:
										self.config_tester.run_common_tests()




						def   lowercase_	(			self							)   ->      Any:
										lowerCAmelCase_	: Tuple      			=			self.model_tester.prepare_config_and_inputs_for_common()
										self.model_tester.check_decoder_model_past_large_inputs(*__lowercase							)







@require_tf
class 			snake_case__(							unittest.TestCase							):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__ :					Any   =							1_000  # show more chars for failing integration tests




						@classmethod
						def   lowercase_	(			cls							)   ->      Optional[Any]:
										lowerCAmelCase_	: Union[str, Any]      			=			f"""Helsinki-NLP/opus-mt-{cls.src}-{cls.tgt}"""
										return cls




						@cached_property
						def   lowercase_	(			self							)   ->      List[str]:
										return AutoTokenizer.from_pretrained(self.model_name							)




						@property
						def   lowercase_	(			self							)   ->      str:
										return self.tokenizer.eos_token_id




						@cached_property
						def   lowercase_	(			self							)   ->      Optional[int]:
										warnings.simplefilter('''error'''							)
										lowerCAmelCase_	: TFMarianMTModel      			=			TFAutoModelForSeqaSeqLM.from_pretrained(self.model_name							)
										assert isinstance(__lowercase				,  __lowercase							)
										lowerCAmelCase_	: Union[str, Any]      			=			model.config
										self.assertListEqual(c.bad_words_ids				,  [[c.pad_token_id]]							)
										self.assertEqual(c.max_length				,  5_1_2							)
										self.assertEqual(c.decoder_start_token_id				,  c.pad_token_id							)
										return model




						def   lowercase_	(			self				,  **__lowercase							)   ->      str:
										lowerCAmelCase_	: int      			=			self.translate_src_text(**__lowercase							)
										self.assertListEqual(self.expected_text				,  __lowercase							)




						def   lowercase_	(			self				,  **__lowercase							)   ->      Optional[int]:
										lowerCAmelCase_	: List[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=1_2_8							)
										lowerCAmelCase_	: Optional[int]      			=			self.tokenizer.batch_decode(generated_ids.numpy()				,  skip_special_tokens=__lowercase							)
										return generated_words







@require_sentencepiece
@require_tokenizers
@require_tf
class 			snake_case__(							UpperCAmelCase__							):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__ :					int   =							'mt'
						SCREAMING_SNAKE_CASE__ :					Union[str, Any]   =							'en'
						SCREAMING_SNAKE_CASE__ :					Optional[Any]   =							['Billi messu b\'mod ġentili, Ġesù fejjaq raġel li kien milqut bil - marda kerha tal - ġdiem.']
						SCREAMING_SNAKE_CASE__ :					Optional[Any]   =							['Touching gently, Jesus healed a man who was affected by the sad disease of leprosy.']




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







@require_sentencepiece
@require_tokenizers
@require_tf
class 			snake_case__(							UpperCAmelCase__							):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__ :					Union[str, Any]   =							'en'
						SCREAMING_SNAKE_CASE__ :					str   =							'zh'
						SCREAMING_SNAKE_CASE__ :					str   =							['My name is Wolfgang and I live in Berlin']
						SCREAMING_SNAKE_CASE__ :					int   =							['我叫沃尔夫冈 我住在柏林']




						@unittest.skip('''Skipping until #12647 is resolved.'''							)
						@slow
						def   lowercase_	(			self							)   ->      Optional[int]:
										self._assert_generated_batch_equal_expected()







@require_sentencepiece
@require_tokenizers
@require_tf
class 			snake_case__(							UpperCAmelCase__							):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__ :					List[Any]   =							'en'
						SCREAMING_SNAKE_CASE__ :					List[str]   =							'ROMANCE'
						SCREAMING_SNAKE_CASE__ :					int   =							[
						    '>>fr<< Don\'t spend so much time watching TV.',
						    '>>pt<< Your message has been sent.',
						    '>>es<< He\'s two years older than me.',
						]
						SCREAMING_SNAKE_CASE__ :					Tuple   =							[
						    '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   lowercase_	(			self							)   ->      List[str]:
										self._assert_generated_batch_equal_expected()




						@unittest.skip('''Skipping until #12647 is resolved.'''							)
						@slow
						def   lowercase_	(			self							)   ->      Dict:
										lowerCAmelCase_	: Optional[int]      			=			TranslationPipeline(self.model				,  self.tokenizer				,  framework='''tf'''							)
										lowerCAmelCase_	: Tuple      			=			pipeline(self.src_text							)
										self.assertEqual(self.expected_text				,  [x['''translation_text'''] for x in output]							)



from typing import List, Optional, Tuple, Union

from . import logging
from .import_utils import is_torch_available, is_torch_version


if is_torch_available():
						import torch

_UpperCAmelCase     :					List[Any]       =logging.get_logger(__name__)  # pylint: disable=invalid-name

try:
						from torch._dynamo import allow_in_graph as maybe_allow_in_graph
except (ImportError, ModuleNotFoundError):



						def      lowerCAmelCase	(						cls	)->	int:
										return cls



def      lowerCAmelCase	(						lowerCAmelCase_	,				lowerCAmelCase_ = None	,				lowerCAmelCase_ = None	,				lowerCAmelCase_ = None	,				lowerCAmelCase_ = None	,				)->	int:
				lowerCAmelCase_	: Tuple      			=			device
				lowerCAmelCase_	: List[str]      			=			shape[0]

				lowerCAmelCase_	: Union[str, Any]      			=			layout or torch.strided
				lowerCAmelCase_	: Union[str, Any]      			=			device or torch.device('''cpu'''	)

				if generator is not None:
								lowerCAmelCase_	: List[str]      			=			generator.device.type if not isinstance(lowerCAmelCase_	,				lowerCAmelCase_	) else generator[0].device.type
								if gen_device_type != device.type and gen_device_type == "cpu":
												lowerCAmelCase_	: Optional[Any]      			=			'''cpu'''
												if device != "mps":
																logger.info(
																    f"""The passed generator was created on 'cpu' even though a tensor on {device} was expected."""
																    f""" Tensors will be created on 'cpu' and then moved to {device}. Note that one can probably"""
																    f""" slighly speed up this function by passing a generator that was created on the {device} device."""	)
								elif gen_device_type != device.type and gen_device_type == "cuda":
												raise ValueError(f"""Cannot generate a {device} tensor from a generator of type {gen_device_type}."""	)

				if isinstance(lowerCAmelCase_	,				lowerCAmelCase_	):
								lowerCAmelCase_	: Dict      			=			(1,) + shape[1:]
								lowerCAmelCase_	: Union[str, Any]      			=			[
								    torch.randn(lowerCAmelCase_	,				generator=generator[i]	,				device=lowerCAmelCase_	,				dtype=lowerCAmelCase_	,				layout=lowerCAmelCase_	)
								    for i in range(lowerCAmelCase_	)
								]
								lowerCAmelCase_	: Tuple      			=			torch.cat(lowerCAmelCase_	,				dim=0	).to(lowerCAmelCase_	)
				else:
								lowerCAmelCase_	: int      			=			torch.randn(lowerCAmelCase_	,				generator=lowerCAmelCase_	,				device=lowerCAmelCase_	,				dtype=lowerCAmelCase_	,				layout=lowerCAmelCase_	).to(lowerCAmelCase_	)

				return latents




def      lowerCAmelCase	(						lowerCAmelCase_	)->	Optional[Any]:
				if is_torch_version('''<'''	,				'''2.0.0'''	) or not hasattr(lowerCAmelCase_	,				'''_dynamo'''	):
								return False
				return isinstance(lowerCAmelCase_	,				torch._dynamo.eval_frame.OptimizedModule	)