

"""simple docstring"""


from __future__ import annotations

import unittest

from transformers import BlenderbotSmallConfig, BlenderbotSmallTokenizer, 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, TFBlenderbotSmallForConditionalGeneration, TFBlenderbotSmallModel





@require_tf
class        snake_case      :
				SCREAMING_SNAKE_CASE_  : Optional[int]   =    BlenderbotSmallConfig
				SCREAMING_SNAKE_CASE_  : int   =    {}
				SCREAMING_SNAKE_CASE_  : str   =    """gelu"""

				def __init__(      self				:							Optional[Any]	,    UpperCamelCase__				:							Any	,    UpperCamelCase__				:							List[str]=1_3	,    UpperCamelCase__				:							List[Any]=7	,    UpperCamelCase__				:							Tuple=True	,    UpperCamelCase__				:							List[Any]=False	,    UpperCamelCase__				:							str=9_9	,    UpperCamelCase__				:							Tuple=3_2	,    UpperCamelCase__				:							Union[str, Any]=2	,    UpperCamelCase__				:							Optional[int]=4	,    UpperCamelCase__				:							List[Any]=3_7	,    UpperCamelCase__				:							int=0.1	,    UpperCamelCase__				:							Dict=0.1	,    UpperCamelCase__				:							Optional[int]=2_0	,    UpperCamelCase__				:							str=2	,    UpperCamelCase__				:							Optional[Any]=1	,    UpperCamelCase__				:							int=0	,    )->  List[Any]:



						'''simple docstring'''


						__lowerCAmelCase:			Dict        =     parent
						__lowerCAmelCase:			Tuple        =     batch_size
						__lowerCAmelCase:			Union[str, Any]        =     seq_length
						__lowerCAmelCase:			List[Any]        =     is_training
						__lowerCAmelCase:			Any        =     use_labels
						__lowerCAmelCase:			Tuple        =     vocab_size
						__lowerCAmelCase:			Tuple        =     hidden_size
						__lowerCAmelCase:			Tuple        =     num_hidden_layers
						__lowerCAmelCase:			List[Any]        =     num_attention_heads
						__lowerCAmelCase:			Any        =     intermediate_size

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

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



						'''simple docstring'''


						__lowerCAmelCase:			List[Any]        =     ids_tensor([self.batch_size, self.seq_length - 1]	,    self.vocab_size)
						__lowerCAmelCase:			Dict        =     tf.expand_dims(tf.constant([self.eos_token_id] * self.batch_size)	,    1)
						__lowerCAmelCase:			Any        =     tf.concat([input_ids, eos_tensor]	,    axis=1)

						__lowerCAmelCase:			Optional[Any]        =     ids_tensor([self.batch_size, self.seq_length]	,    self.vocab_size)

						__lowerCAmelCase:			Optional[Any]        =     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:			Any        =     prepare_blenderbot_small_inputs_dict(__lowercase	,    __lowercase	,    __lowercase)
						return config, inputs_dict




				def 				lowercase_	(      self				:							Dict	,    UpperCamelCase__				:							str	,    UpperCamelCase__				:							Optional[int])->  Optional[int]:



						'''simple docstring'''


						__lowerCAmelCase:			str        =     TFBlenderbotSmallModel(config=__lowercase).get_decoder()
						__lowerCAmelCase:			Dict        =     inputs_dict["input_ids"]

						__lowerCAmelCase:			Any        =     input_ids[:1, :]
						__lowerCAmelCase:			Tuple        =     inputs_dict["attention_mask"][:1, :]
						__lowerCAmelCase:			Union[str, Any]        =     inputs_dict["head_mask"]
						__lowerCAmelCase:			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:			Any        =     ids_tensor((self.batch_size, 3)	,    config.vocab_size)
						__lowerCAmelCase:			Tuple        =     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:			int        =     tf.concat([attention_mask, next_attn_mask]	,    axis=-1)

						__lowerCAmelCase:			Union[str, Any]        =     model(__lowercase	,    attention_mask=__lowercase)[0]
						__lowerCAmelCase:			str        =     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:			Dict        =     int(ids_tensor((1,)	,    output_from_past.shape[-1]))
						__lowerCAmelCase:			Tuple        =     output_from_no_past[:, -3:, random_slice_idx]
						__lowerCAmelCase:			Dict        =     output_from_past[:, :, random_slice_idx]

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






def        a__       (       __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   ,					)      ->      List[str]:





		if attention_mask is None:
				__lowerCAmelCase:			Tuple        =     tf.cast(tf.math.not_equal(lowercase__   ,					config.pad_token_id					)   ,					tf.inta					)
		if decoder_attention_mask is None:
				__lowerCAmelCase:			Optional[Any]        =     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:			Union[str, Any]        =     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      (							__snake_case,				__snake_case,				unittest.TestCase    ):
				SCREAMING_SNAKE_CASE_  : List[Any]   =    (
				    (TFBlenderbotSmallForConditionalGeneration, TFBlenderbotSmallModel) if is_tf_available() else ()
				)
				SCREAMING_SNAKE_CASE_  : Any   =    (TFBlenderbotSmallForConditionalGeneration,) if is_tf_available() else ()
				SCREAMING_SNAKE_CASE_  : List[Any]   =    (
				    {
				        """conversational""": TFBlenderbotSmallForConditionalGeneration,
				        """feature-extraction""": TFBlenderbotSmallModel,
				        """summarization""": TFBlenderbotSmallForConditionalGeneration,
				        """text2text-generation""": TFBlenderbotSmallForConditionalGeneration,
				        """translation""": TFBlenderbotSmallForConditionalGeneration,
				    }
				    if is_tf_available()
				    else {}
				)
				SCREAMING_SNAKE_CASE_  : Dict   =    True
				SCREAMING_SNAKE_CASE_  : Tuple   =    False
				SCREAMING_SNAKE_CASE_  : Union[str, Any]   =    False

				def 				lowercase_	(      self				:							Optional[int])->  Tuple:



						'''simple docstring'''


						__lowerCAmelCase:			str        =     TFBlenderbotSmallModelTester(self)
						__lowerCAmelCase:			Any        =     ConfigTester(self	,    config_class=__lowercase)

				def 				lowercase_	(      self				:							Optional[Any])->  List[str]:



						'''simple docstring'''


						self.config_tester.run_common_tests()




				def 				lowercase_	(      self				:							Optional[Any])->  int:



						'''simple docstring'''


						__lowerCAmelCase:			str        =     self.model_tester.prepare_config_and_inputs_for_common()
						self.model_tester.check_decoder_model_past_large_inputs(*__lowercase)





@require_tokenizers
@require_tf
class        snake_case      (							unittest.TestCase    ):
				SCREAMING_SNAKE_CASE_  : List[str]   =    [
				    """Social anxiety\nWow, I am never shy. Do you have anxiety?\nYes. I end up sweating and blushing and feel like """
				    """  i'm going to throw up.\nand why is that?"""
				]
				SCREAMING_SNAKE_CASE_  : int   =    """facebook/blenderbot_small-90M"""

				@cached_property
				def 				lowercase_	(      self				:							Dict)->  Tuple:



						'''simple docstring'''


						return BlenderbotSmallTokenizer.from_pretrained("facebook/blenderbot-90M")

				@cached_property
				def 				lowercase_	(      self				:							Any)->  Tuple:



						'''simple docstring'''


						__lowerCAmelCase:			str        =     TFAutoModelForSeqaSeqLM.from_pretrained(self.model_name)
						return model




				@slow
				def 				lowercase_	(      self				:							Dict)->  Dict:



						'''simple docstring'''


						__lowerCAmelCase:			int        =     self.tokenizer(self.src_text	,    return_tensors="tf")
						__lowerCAmelCase:			Dict        =     self.model.generate(
						    model_inputs.input_ids	,    attention_mask=model_inputs.attention_mask	,    num_beams=2	,    use_cache=__lowercase	,    )
						__lowerCAmelCase:			Dict        =     self.tokenizer.batch_decode(generated_ids.numpy()	,    skip_special_tokens=__lowercase)[0]
						assert generated_words in (
						    "i don't know. i just feel like i'm going to throw up. it's not fun.",
						    "i'm not sure. i just feel like i've been feeling like i have to be in a certain place",
						    "i'm not sure. i just feel like i've been in a bad situation.",
						)





'''simple docstring'''

import enum
import warnings

from ..tokenization_utils import TruncationStrategy
from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
       import tensorflow as tf

       from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING

if is_torch_available():
       from ..models.auto.modeling_auto import MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING

UpperCAmelCase					 =	logging.get_logger(__name__)


class 	lowerCAmelCase  (   enum.Enum ):
       lowerCAmelCase_												=			0
       lowerCAmelCase_												=			1






@add_end_docstrings(A )
class 	lowerCAmelCase  (   A ):
       lowerCAmelCase_												=			"generated"







       def __init__(				self      :			Tuple    ,   *__lowercase      :			Tuple    ,   **__lowercase      :			Optional[int]   ):







            """simple docstring"""







            super().__init__(*__lowercase    ,   **__lowercase   )

            self.check_model_type(
                TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
                if self.framework == 'tf'
                else MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING   )







       def       snake_case (				self      :			Dict    ,   __lowercase      :			Any=None    ,   __lowercase      :			int=None    ,   __lowercase      :			List[str]=None    ,   __lowercase      :			int=None    ,   __lowercase      :			str=None    ,   __lowercase      :			List[Any]=None    ,   **__lowercase      :			Union[str, Any]    ,   ):







            """simple docstring"""







            __lowercase      		={}
            if truncation is not None:
                 __lowercase      		=truncation

            __lowercase      		=generate_kwargs

            __lowercase      		={}
            if return_tensors is not None and return_type is None:
                 __lowercase      		=ReturnType.TENSORS if return_tensors else ReturnType.TEXT
            if return_type is not None:
                 __lowercase      		=return_type

            if clean_up_tokenization_spaces is not None:
                 __lowercase      		=clean_up_tokenization_spaces

            if stop_sequence is not None:
                 __lowercase      		=self.tokenizer.encode(__lowercase    ,   add_special_tokens=__lowercase   )
                 if len(__lowercase   ) > 1:
                      warnings.warn(
                          'Stopping on a multiple token sequence is not yet supported on transformers. The first token of'
                          ' the stop sequence will be used as the stop sequence string in the interim.'   )
                 __lowercase      		=stop_sequence_ids[0]

            return preprocess_params, forward_params, postprocess_params







       def       snake_case (				self      :			List[Any]    ,   __lowercase      :			int    ,   __lowercase      :			int    ,   __lowercase      :			int   ):







            """simple docstring"""







            return True







       def       snake_case (				self      :			Optional[Any]    ,   *__lowercase      :			Optional[int]    ,   __lowercase      :			Union[str, Any]   ):







            """simple docstring"""







            __lowercase      		=self.model.config.prefix if self.model.config.prefix is not None else ''
            if isinstance(args[0]    ,   __lowercase   ):
                 if self.tokenizer.pad_token_id is None:
                      raise ValueError('Please make sure that the tokenizer has a pad_token_id when using a batch input'   )
                 __lowercase      		=([prefix + arg for arg in args[0]],)
                 __lowercase      		=True

            elif isinstance(args[0]    ,   __lowercase   ):
                 __lowercase      		=(prefix + args[0],)
                 __lowercase      		=False
            else:
                 raise ValueError(
                     f''' `args[0]`: {args[0]} have the wrong format. The should be either of type `str` or type `list`'''   )
            __lowercase      		=self.tokenizer(*__lowercase    ,   padding=__lowercase    ,   truncation=__lowercase    ,   return_tensors=self.framework   )
            # This is produced by tokenizers but is an invalid generate kwargs
            if "token_type_ids" in inputs:
                 del inputs["token_type_ids"]
            return inputs







       def __call__(				self      :			Any    ,   *__lowercase      :			int    ,   **__lowercase      :			Optional[Any]   ):







            """simple docstring"""







            __lowercase      		=super().__call__(*__lowercase    ,   **__lowercase   )
            if (
                isinstance(args[0]    ,   __lowercase   )
                and all(isinstance(__lowercase    ,   __lowercase   ) for el in args[0]   )
                and all(len(__lowercase   ) == 1 for res in result   )
            ):
                 return [res[0] for res in result]
            return result







       def       snake_case (				self      :			List[str]    ,   __lowercase      :			Tuple    ,   __lowercase      :			List[Any]=TruncationStrategy.DO_NOT_TRUNCATE    ,   **__lowercase      :			Dict   ):







            """simple docstring"""







            __lowercase      		=self._parse_and_tokenize(__lowercase    ,   truncation=__lowercase    ,   **__lowercase   )
            return inputs







       def       snake_case (				self      :			Optional[int]    ,   __lowercase      :			Tuple    ,   **__lowercase      :			Tuple   ):







            """simple docstring"""







            if self.framework == "pt":
                 __lowercase		,  __lowercase      		=model_inputs['input_ids'].shape
            elif self.framework == "tf":
                 __lowercase		,  __lowercase      		=tf.shape(model_inputs['input_ids']   ).numpy()

            __lowercase      		=generate_kwargs.get('min_length'    ,   self.model.config.min_length   )
            __lowercase      		=generate_kwargs.get('max_length'    ,   self.model.config.max_length   )
            self.check_inputs(__lowercase    ,   generate_kwargs['min_length']    ,   generate_kwargs['max_length']   )
            __lowercase      		=self.model.generate(**__lowercase    ,   **__lowercase   )
            __lowercase      		=output_ids.shape[0]
            if self.framework == "pt":
                 __lowercase      		=output_ids.reshape(__lowercase    ,   out_b // in_b    ,   *output_ids.shape[1:]   )
            elif self.framework == "tf":
                 __lowercase      		=tf.reshape(__lowercase    ,   (in_b, out_b // in_b, *output_ids.shape[1:])   )
            return {"output_ids": output_ids}







       def       snake_case (				self      :			str    ,   __lowercase      :			Optional[Any]    ,   __lowercase      :			Optional[int]=ReturnType.TEXT    ,   __lowercase      :			Union[str, Any]=False   ):







            """simple docstring"""







            __lowercase      		=[]
            for output_ids in model_outputs["output_ids"][0]:
                 if return_type == ReturnType.TENSORS:
                      __lowercase      		={f'''{self.return_name}_token_ids''': output_ids}
                 elif return_type == ReturnType.TEXT:
                      __lowercase      		={
                          f'''{self.return_name}_text''': self.tokenizer.decode(
                              __lowercase    ,   skip_special_tokens=__lowercase    ,   clean_up_tokenization_spaces=__lowercase    ,   )
                      }
                 records.append(__lowercase   )
            return records






@add_end_docstrings(A )
class 	lowerCAmelCase  (   A ):
       lowerCAmelCase_												=			"summary"







       def __call__(				self      :			Dict    ,   *__lowercase      :			Dict    ,   **__lowercase      :			Dict   ):







            """simple docstring"""







            return super().__call__(*__lowercase    ,   **__lowercase   )







       def       snake_case (				self      :			str    ,   __lowercase      :			int    ,   __lowercase      :			int    ,   __lowercase      :			int   ):







            """simple docstring"""







            if max_length < min_length:
                 logger.warning(f'''Your min_length={min_length} must be inferior than your max_length={max_length}.'''   )

            if input_length < max_length:
                 logger.warning(
                     f'''Your max_length is set to {max_length}, but your input_length is only {input_length}. Since this is '''
                     'a summarization task, where outputs shorter than the input are typically wanted, you might '
                     f'''consider decreasing max_length manually, e.g. summarizer(\'...\', max_length={input_length//2})'''   )






@add_end_docstrings(A )
class 	lowerCAmelCase  (   A ):
       lowerCAmelCase_												=			"translation"







       def       snake_case (				self      :			int    ,   __lowercase      :			int    ,   __lowercase      :			int    ,   __lowercase      :			int   ):







            """simple docstring"""







            if input_length > 0.9 * max_length:
                 logger.warning(
                     f'''Your input_length: {input_length} is bigger than 0.9 * max_length: {max_length}. You might consider '''
                     'increasing your max_length manually, e.g. translator(\'...\', max_length=400)'   )
            return True







       def       snake_case (				self      :			int    ,   *__lowercase      :			str    ,   __lowercase      :			List[str]=TruncationStrategy.DO_NOT_TRUNCATE    ,   __lowercase      :			List[Any]=None    ,   __lowercase      :			Optional[int]=None   ):







            """simple docstring"""







            if getattr(self.tokenizer    ,   '_build_translation_inputs'    ,   __lowercase   ):
                 return self.tokenizer._build_translation_inputs(
                     *__lowercase    ,   return_tensors=self.framework    ,   truncation=__lowercase    ,   src_lang=__lowercase    ,   tgt_lang=__lowercase   )
            else:
                 return super()._parse_and_tokenize(*__lowercase    ,   truncation=__lowercase   )







       def       snake_case (				self      :			Optional[int]    ,   __lowercase      :			List[str]=None    ,   __lowercase      :			Optional[Any]=None    ,   **__lowercase      :			int   ):







            """simple docstring"""







            __lowercase		,  __lowercase		,  __lowercase      		=super()._sanitize_parameters(**__lowercase   )
            if src_lang is not None:
                 __lowercase      		=src_lang
            if tgt_lang is not None:
                 __lowercase      		=tgt_lang
            if src_lang is None and tgt_lang is None:
                 # Backward compatibility, direct arguments use is preferred.
                 __lowercase      		=kwargs.get('task'    ,   self.task   )
                 __lowercase      		=task.split('_'   )
                 if task and len(__lowercase   ) == 4:
                      # translation, XX, to YY
                      __lowercase      		=items[1]
                      __lowercase      		=items[3]
            return preprocess_params, forward_params, postprocess_params







       def __call__(				self      :			Optional[Any]    ,   *__lowercase      :			Any    ,   **__lowercase      :			List[Any]   ):







            """simple docstring"""







            return super().__call__(*__lowercase    ,   **__lowercase   )


