
from __future__ import annotations

import unittest

from transformers import BlenderbotConfig, BlenderbotTokenizer, is_tf_available
from transformers.testing_utils import 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, TFBlenderbotForConditionalGeneration, TFBlenderbotModel







@require_tf
class        lowercase	:
       lowercase_ : Any       =BlenderbotConfig
       lowercase_ : int       ={}
       lowercase_ : Optional[Any]       ='''gelu'''




       def __init__(       self     ,A__     ,A__=1_3     ,A__=7     ,A__=True     ,A__=False     ,A__=9_9     ,A__=3_2     ,A__=2     ,A__=4     ,A__=3_7     ,A__=0.1     ,A__=0.1     ,A__=2_0     ,A__=2     ,A__=1     ,A__=0     ,):
            lowercase					=		parent
            lowercase					=		batch_size
            lowercase					=		seq_length
            lowercase					=		is_training
            lowercase					=		use_labels
            lowercase					=		vocab_size
            lowercase					=		hidden_size
            lowercase					=		num_hidden_layers
            lowercase					=		num_attention_heads
            lowercase					=		intermediate_size

            lowercase					=		hidden_dropout_prob
            lowercase					=		attention_probs_dropout_prob
            lowercase					=		max_position_embeddings
            lowercase					=		eos_token_id
            lowercase					=		pad_token_id
            lowercase					=		bos_token_id




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

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

            lowercase					=		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     ,)
            lowercase					=		prepare_blenderbot_inputs_dict(__UpperCamelCase     ,__UpperCamelCase     ,__UpperCamelCase)
            return config, inputs_dict





       def 				A__   (       self     ,A__     ,A__):
            lowercase					=		TFBlenderbotModel(config=__UpperCamelCase).get_decoder()
            lowercase					=		inputs_dict['''input_ids''']

            lowercase					=		input_ids[:1, :]
            lowercase					=		inputs_dict['''attention_mask'''][:1, :]
            lowercase					=		inputs_dict['''head_mask''']
            lowercase					=		1

            # first forward pass
            lowercase					=		model(__UpperCamelCase     ,attention_mask=__UpperCamelCase     ,head_mask=__UpperCamelCase     ,use_cache=__UpperCamelCase)

            lowercase			,		lowercase					=		outputs.to_tuple()

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

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

            lowercase					=		model(__UpperCamelCase     ,attention_mask=__UpperCamelCase)[0]
            lowercase					=		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
            lowercase					=		int(ids_tensor((1,)     ,output_from_past.shape[-1]))
            lowercase					=		output_from_no_past[:, -3:, random_slice_idx]
            lowercase					=		output_from_past[:, :, random_slice_idx]

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



def  UpperCamelCase					(							lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__=None				,    lowerCAmelCase__=None				,    lowerCAmelCase__=None				,    lowerCAmelCase__=None				,    lowerCAmelCase__=None				,    ):
     '''simple docstring'''
     if attention_mask is None:
          lowercase					=		tf.cast(tf.math.not_equal(UpperCAmelCase__				,    config.pad_token_id	)				,    tf.inta	)
     if decoder_attention_mask is None:
          lowercase					=		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:
          lowercase					=		tf.ones((config.encoder_layers, config.encoder_attention_heads)	)
     if decoder_head_mask is None:
          lowercase					=		tf.ones((config.decoder_layers, config.decoder_attention_heads)	)
     if cross_attn_head_mask is None:
          lowercase					=		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        lowercase	(      SCREAMING_SNAKE_CASE__ ,						SCREAMING_SNAKE_CASE__ ,						unittest.TestCase	):
       lowercase_ : Any       =(TFBlenderbotForConditionalGeneration, TFBlenderbotModel) if is_tf_available() else ()
       lowercase_ : str       =(TFBlenderbotForConditionalGeneration,) if is_tf_available() else ()
       lowercase_ : Any       =(
           {
               '''conversational''': TFBlenderbotForConditionalGeneration,
               '''feature-extraction''': TFBlenderbotModel,
               '''summarization''': TFBlenderbotForConditionalGeneration,
               '''text2text-generation''': TFBlenderbotForConditionalGeneration,
               '''translation''': TFBlenderbotForConditionalGeneration,
           }
           if is_tf_available()
           else {}
       )
       lowercase_ : Optional[int]       =True
       lowercase_ : Dict       =False
       lowercase_ : int       =False




       def 				A__   (       self):
            lowercase					=		TFBlenderbotModelTester(self)
            lowercase					=		ConfigTester(self     ,config_class=__UpperCamelCase)




       def 				A__   (       self):
            self.config_tester.run_common_tests()





       def 				A__   (       self):
            lowercase					=		self.model_tester.prepare_config_and_inputs_for_common()
            self.model_tester.check_decoder_model_past_large_inputs(*__UpperCamelCase)







@require_tokenizers
@require_tf
class        lowercase	(      unittest.TestCase	):
       lowercase_ : List[Any]       =['''My friends are cool but they eat too many carbs.''']
       lowercase_ : Optional[int]       ='''facebook/blenderbot-400M-distill'''




       @cached_property
       def 				A__   (       self):
            return BlenderbotTokenizer.from_pretrained(self.model_name)




       @cached_property
       def 				A__   (       self):
            lowercase					=		TFAutoModelForSeqaSeqLM.from_pretrained(self.model_name)
            return model





       @slow
       def 				A__   (       self):
            lowercase					=		self.tokenizer(self.src_text     ,return_tensors='''tf''')
            lowercase					=		self.model.generate(
                model_inputs.input_ids     ,)
            lowercase					=		self.tokenizer.batch_decode(generated_ids.numpy()     ,skip_special_tokens=__UpperCamelCase)[0]
            assert (
                generated_words
                == " That's unfortunate. Are they trying to lose weight or are they just trying to be healthier?"
            )






import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
							import torch_xla.core.xla_model as xm
							import torch_xla.debug.metrics as met







class        lowercase	(      SCREAMING_SNAKE_CASE__	):




							def __init__(       self     ,*A__     ,A__=None     ,A__=None     ,**A__):
												super().__init__(*A__     ,**A__)
												lowercase					=		eval_examples
												lowercase					=		post_process_function




							def 				A__   (       self     ,A__ = None     ,A__=None     ,A__ = None     ,A__ = "eval"     ,**A__     ,):
												lowercase					=		gen_kwargs.copy()
												lowercase					=		(
												    gen_kwargs['''max_length'''] if gen_kwargs.get('''max_length''') is not None else self.args.generation_max_length
												)
												lowercase					=		(
												    gen_kwargs['''num_beams'''] if gen_kwargs.get('''num_beams''') is not None else self.args.generation_num_beams
												)
												lowercase					=		gen_kwargs

												lowercase					=		self.eval_dataset if eval_dataset is None else eval_dataset
												lowercase					=		self.get_eval_dataloader(A__)
												lowercase					=		self.eval_examples if eval_examples is None else eval_examples

												# Temporarily disable metric computation, we will do it in the loop here.
												lowercase					=		self.compute_metrics
												lowercase					=		None
												lowercase					=		time.time()
												lowercase					=		self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
												try:
																	lowercase					=		eval_loop(
																	    A__     ,description='''Evaluation'''     ,prediction_loss_only=True if compute_metrics is None else None     ,ignore_keys=A__     ,metric_key_prefix=A__     ,)
												finally:
																	lowercase					=		compute_metrics
												lowercase					=		self.args.eval_batch_size * self.args.world_size
												if f'{metric_key_prefix}_jit_compilation_time' in output.metrics:
																	start_time += output.metrics[f'{metric_key_prefix}_jit_compilation_time']
												output.metrics.update(
												    speed_metrics(
												        A__     ,A__     ,num_samples=output.num_samples     ,num_steps=math.ceil(output.num_samples / total_batch_size)     ,))

												if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
																	# Only the main node write the results by default
																	lowercase					=		self.post_process_function(A__     ,A__     ,A__)
																	lowercase					=		self.compute_metrics(A__)

																	# Prefix all keys with metric_key_prefix + '_'
																	for key in list(metrics.keys()):
																						if not key.startswith(f'{metric_key_prefix}_'):
																											lowercase					=		metrics.pop(A__)

																	metrics.update(output.metrics)
												else:
																	lowercase					=		output.metrics

												if self.args.should_log:
																	# Only the main node log the results by default
																	self.log(A__)

												if self.args.tpu_metrics_debug or self.args.debug:
																	# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
																	xm.master_print(met.metrics_report())

												lowercase					=		self.callback_handler.on_evaluate(self.args     ,self.state     ,self.control     ,A__)
												return metrics





							def 				A__   (       self     ,A__     ,A__     ,A__=None     ,A__ = "test"     ,**A__):
												lowercase					=		gen_kwargs.copy()

												lowercase					=		self.get_test_dataloader(A__)

												# Temporarily disable metric computation, we will do it in the loop here.
												lowercase					=		self.compute_metrics
												lowercase					=		None
												lowercase					=		time.time()
												lowercase					=		self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
												try:
																	lowercase					=		eval_loop(
																	    A__     ,description='''Prediction'''     ,prediction_loss_only=True if compute_metrics is None else None     ,ignore_keys=A__     ,metric_key_prefix=A__     ,)
												finally:
																	lowercase					=		compute_metrics

												lowercase					=		self.args.eval_batch_size * self.args.world_size
												if f'{metric_key_prefix}_jit_compilation_time' in output.metrics:
																	start_time += output.metrics[f'{metric_key_prefix}_jit_compilation_time']
												output.metrics.update(
												    speed_metrics(
												        A__     ,A__     ,num_samples=output.num_samples     ,num_steps=math.ceil(output.num_samples / total_batch_size)     ,))
												if self.post_process_function is None or self.compute_metrics is None:
																	return output

												lowercase					=		self.post_process_function(A__     ,A__     ,A__     ,'''predict''')
												lowercase					=		self.compute_metrics(A__)

												# Prefix all keys with metric_key_prefix + '_'
												for key in list(metrics.keys()):
																	if not key.startswith(f'{metric_key_prefix}_'):
																						lowercase					=		metrics.pop(A__)
												metrics.update(output.metrics)
												return PredictionOutput(predictions=predictions.predictions     ,label_ids=predictions.label_ids     ,metrics=A__)




