





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  lowerCAmelCase							:
  __lowerCamelCase		=      BlenderbotConfig
  __lowerCamelCase		=      {}
  __lowerCamelCase		=      'gelu'







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






     '''simple docstring'''





     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   UpperCAmelCase					(      self  :Optional[int] ):






     '''simple docstring'''





     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(_lowercase   ,							_lowercase   ,							_lowercase )
     return config, inputs_dict







  def   UpperCAmelCase					(      self  :Tuple   ,							_lowercase  :Optional[int]   ,							_lowercase  :Union[str, Any] ):






     '''simple docstring'''





     lowercase__			       =  TFBlenderbotModel(config=_lowercase ).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(_lowercase   ,							attention_mask=_lowercase   ,							head_mask=_lowercase   ,							use_cache=_lowercase )

     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(_lowercase   ,							attention_mask=_lowercase )[0]
     lowercase__			       =  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
     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(_lowercase   ,							_lowercase   ,							rtol=1e-3 )


def _A       (   __magic_name__  , __magic_name__  , __magic_name__  , __magic_name__=None  , __magic_name__=None  , __magic_name__=None  , __magic_name__=None  , __magic_name__=None  , ):
   if attention_mask is None:
      lowercase__			       =  tf.cast(tf.math.not_equal(__magic_name__  , 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  lowerCAmelCase							(       lowercase_							,							lowercase_							,							unittest.TestCase ):
  __lowerCamelCase		=      (TFBlenderbotForConditionalGeneration, TFBlenderbotModel) if is_tf_available() else ()
  __lowerCamelCase		=      (TFBlenderbotForConditionalGeneration,) if is_tf_available() else ()
  __lowerCamelCase		=      (
      {
          'conversational': TFBlenderbotForConditionalGeneration,
          'feature-extraction': TFBlenderbotModel,
          'summarization': TFBlenderbotForConditionalGeneration,
          'text2text-generation': TFBlenderbotForConditionalGeneration,
          'translation': TFBlenderbotForConditionalGeneration,
      }
      if is_tf_available()
      else {}
  )
  __lowerCamelCase		=      True
  __lowerCamelCase		=      False
  __lowerCamelCase		=      False







  def   UpperCAmelCase					(      self  :Any ):






     '''simple docstring'''





     lowercase__			       =  TFBlenderbotModelTester(self )
     lowercase__			       =  ConfigTester(self   ,							config_class=_lowercase )







  def   UpperCAmelCase					(      self  :Tuple ):






     '''simple docstring'''





     self.config_tester.run_common_tests()







  def   UpperCAmelCase					(      self  :int ):






     '''simple docstring'''





     lowercase__			       =  self.model_tester.prepare_config_and_inputs_for_common()
     self.model_tester.check_decoder_model_past_large_inputs(*_lowercase )





@require_tokenizers
@require_tf
class  lowerCAmelCase							(       unittest.TestCase ):
  __lowerCamelCase		=      ['My friends are cool but they eat too many carbs.']
  __lowerCamelCase		=      'facebook/blenderbot-400M-distill'







  @cached_property
  def   UpperCAmelCase					(      self  :Optional[Any] ):






     '''simple docstring'''





     return BlenderbotTokenizer.from_pretrained(self.model_name )







  @cached_property
  def   UpperCAmelCase					(      self  :List[str] ):






     '''simple docstring'''





     lowercase__			       =  TFAutoModelForSeqaSeqLM.from_pretrained(self.model_name )
     return model







  @slow
  def   UpperCAmelCase					(      self  :List[Any] ):






     '''simple docstring'''





     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=_lowercase )[0]
     assert (
         generated_words
         == " That's unfortunate. Are they trying to lose weight or are they just trying to be healthier?"
     )













import math
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from .scheduling_utils import SchedulerMixin, SchedulerOutput





class  lowerCAmelCase							(       lowercase_							,							lowercase_ ):
  __lowerCamelCase		=      1







  @register_to_config
  def __init__(      self  :Dict   ,							_lowercase  :int = 10_00   ,							_lowercase  :Optional[Union[np.ndarray, List[float]]] = None ):






     '''simple docstring'''





     self.set_timesteps(_lowercase )

     # standard deviation of the initial noise distribution
     lowercase__			       =  1.0

     # For now we only support F-PNDM, i.e. the runge-kutta method
     # For more information on the algorithm please take a look at the paper: https://arxiv.org/pdf/2202.09778.pdf
     # mainly at formula (9), (12), (13) and the Algorithm 2.
     lowercase__			       =  4

     # running values
     lowercase__			       =  []







  def   UpperCAmelCase					(      self  :str   ,							_lowercase  :int   ,							_lowercase  :Union[str, torch.device] = None ):






     '''simple docstring'''





     lowercase__			       =  num_inference_steps
     lowercase__			       =  torch.linspace(1   ,							0   ,							num_inference_steps + 1 )[:-1]
     lowercase__			       =  torch.cat([steps, torch.tensor([0.0] )] )

     if self.config.trained_betas is not None:
        lowercase__			       =  torch.tensor(self.config.trained_betas   ,							dtype=torch.floataa )
     else:
        lowercase__			       =  torch.sin(steps * math.pi / 2 ) ** 2

     lowercase__			       =  (1.0 - self.betas**2) ** 0.5

     lowercase__			       =  (torch.atana(self.betas   ,							self.alphas ) / math.pi * 2)[:-1]
     lowercase__			       =  timesteps.to(_lowercase )

     lowercase__			       =  []







  def   UpperCAmelCase					(      self  :Optional[int]   ,							_lowercase  :torch.FloatTensor   ,							_lowercase  :int   ,							_lowercase  :torch.FloatTensor   ,							_lowercase  :bool = True   ,							):






     '''simple docstring'''





     if self.num_inference_steps is None:
        raise ValueError(
            "Number of inference steps is 'None', you need to run 'set_timesteps' after creating the scheduler" )

     lowercase__			       =  (self.timesteps == timestep).nonzero().item()
     lowercase__			       =  timestep_index + 1

     lowercase__			       =  sample * self.betas[timestep_index] + model_output * self.alphas[timestep_index]
     self.ets.append(_lowercase )

     if len(self.ets ) == 1:
        lowercase__			       =  self.ets[-1]
     elif len(self.ets ) == 2:
        lowercase__			       =  (3 * self.ets[-1] - self.ets[-2]) / 2
     elif len(self.ets ) == 3:
        lowercase__			       =  (23 * self.ets[-1] - 16 * self.ets[-2] + 5 * self.ets[-3]) / 12
     else:
        lowercase__			       =  (1 / 24) * (55 * self.ets[-1] - 59 * self.ets[-2] + 37 * self.ets[-3] - 9 * self.ets[-4])

     lowercase__			       =  self._get_prev_sample(_lowercase   ,							_lowercase   ,							_lowercase   ,							_lowercase )

     if not return_dict:
        return (prev_sample,)

     return SchedulerOutput(prev_sample=_lowercase )







  def   UpperCAmelCase					(      self  :Any   ,							_lowercase  :torch.FloatTensor   ,							*_lowercase  :int   ,							**_lowercase  :int ):






     '''simple docstring'''





     return sample







  def   UpperCAmelCase					(      self  :str   ,							_lowercase  :Tuple   ,							_lowercase  :int   ,							_lowercase  :Optional[Any]   ,							_lowercase  :List[str] ):






     '''simple docstring'''





     lowercase__			       =  self.alphas[timestep_index]
     lowercase__			       =  self.betas[timestep_index]

     lowercase__			       =  self.alphas[prev_timestep_index]
     lowercase__			       =  self.betas[prev_timestep_index]

     lowercase__			       =  (sample - sigma * ets) / max(_lowercase   ,							1e-8 )
     lowercase__			       =  next_alpha * pred + ets * next_sigma

     return prev_sample







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






     '''simple docstring'''





     return self.config.num_train_timesteps






