


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


			'''simple docstring'''







			lowercase					:	Dict									=BlenderbotConfig
			lowercase					:	Dict									={}
			lowercase					:	Dict									="""gelu"""
			def __init__(     self   ,							UpperCamelCase_   ,							UpperCamelCase_=13   ,							UpperCamelCase_=7   ,							UpperCamelCase_=True   ,							UpperCamelCase_=False   ,							UpperCamelCase_=99   ,							UpperCamelCase_=32   ,							UpperCamelCase_=2   ,							UpperCamelCase_=4   ,							UpperCamelCase_=37   ,							UpperCamelCase_=0.1   ,							UpperCamelCase_=0.1   ,							UpperCamelCase_=20   ,							UpperCamelCase_=2   ,							UpperCamelCase_=1   ,							UpperCamelCase_=0   ,							):
								lowercase_       :Union[str, Any]     =       parent
								lowercase_       :Any     =       batch_size
								lowercase_       :int     =       seq_length
								lowercase_       :Any     =       is_training
								lowercase_       :Union[str, Any]     =       use_labels
								lowercase_       :Dict     =       vocab_size
								lowercase_       :str     =       hidden_size
								lowercase_       :List[str]     =       num_hidden_layers
								lowercase_       :Optional[Any]     =       num_attention_heads
								lowercase_       :List[Any]     =       intermediate_size

								lowercase_       :Optional[int]     =       hidden_dropout_prob
								lowercase_       :List[Any]     =       attention_probs_dropout_prob
								lowercase_       :Optional[int]     =       max_position_embeddings
								lowercase_       :Optional[int]     =       eos_token_id
								lowercase_       :str     =       pad_token_id
								lowercase_       :Optional[Any]     =       bos_token_id
			def 		UpperCamelCase  (     self		):
								lowercase_       :Optional[Any]     =       ids_tensor([self.batch_size, self.seq_length - 1]   ,							self.vocab_size		)
								lowercase_       :Dict     =       tf.expand_dims(tf.constant([self.eos_token_id] * self.batch_size		)   ,							1		)
								lowercase_       :Dict     =       tf.concat([input_ids, eos_tensor]   ,							axis=1		)

								lowercase_       :Union[str, Any]     =       ids_tensor([self.batch_size, self.seq_length]   ,							self.vocab_size		)

								lowercase_       :Optional[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   ,							)
								lowercase_       :List[Any]     =       prepare_blenderbot_inputs_dict(UpperCAmelCase_   ,							UpperCAmelCase_   ,							UpperCAmelCase_		)
								return config, inputs_dict




			def 		UpperCamelCase  (     self   ,							UpperCamelCase_   ,							UpperCamelCase_		):
								lowercase_       :Any     =       TFBlenderbotModel(config=UpperCAmelCase_		).get_decoder()
								lowercase_       :Dict     =       inputs_dict['''input_ids''']

								lowercase_       :int     =       input_ids[:1, :]
								lowercase_       :Optional[int]     =       inputs_dict['''attention_mask'''][:1, :]
								lowercase_       :Optional[int]     =       inputs_dict['''head_mask''']
								lowercase_       :List[str]     =       1

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

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

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

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

								lowercase_       :str     =       model(UpperCAmelCase_   ,							attention_mask=UpperCAmelCase_		)[0]
								lowercase_       :List[str]     =       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_       :List[str]     =       int(ids_tensor((1,)   ,							output_from_past.shape[-1]		)		)
								lowercase_       :List[str]     =       output_from_no_past[:, -3:, random_slice_idx]
								lowercase_       :List[Any]     =       output_from_past[:, :, random_slice_idx]

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




def     UpperCamelCase    (      _a       ,		_a       ,		_a       ,		_a=None       ,		_a=None       ,		_a=None       ,		_a=None       ,		_a=None       ,		)					->							Tuple:
					'''simple docstring'''





					if attention_mask is None:
										lowercase_       :List[str]     =       tf.cast(tf.math.not_equal(_snake_case       ,		config.pad_token_id		)       ,		tf.inta		)
					if decoder_attention_mask is None:
										lowercase_       :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:
										lowercase_       :Optional[Any]     =       tf.ones((config.encoder_layers, config.encoder_attention_heads)		)
					if decoder_head_mask is None:
										lowercase_       :Any     =       tf.ones((config.decoder_layers, config.decoder_attention_heads)		)
					if cross_attn_head_mask is None:
										lowercase_       :Dict     =       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      UpperCamelCase					(		__UpperCAmelCase						,      __UpperCAmelCase						,      unittest.TestCase     ):


			'''simple docstring'''







			lowercase					:	Union[str, Any]									=(TFBlenderbotForConditionalGeneration, TFBlenderbotModel) if is_tf_available() else ()
			lowercase					:	Union[str, Any]									=(TFBlenderbotForConditionalGeneration,) if is_tf_available() else ()
			lowercase					:	int									=(
			    {
			        """conversational""": TFBlenderbotForConditionalGeneration,
			        """feature-extraction""": TFBlenderbotModel,
			        """summarization""": TFBlenderbotForConditionalGeneration,
			        """text2text-generation""": TFBlenderbotForConditionalGeneration,
			        """translation""": TFBlenderbotForConditionalGeneration,
			    }
			    if is_tf_available()
			    else {}
			)
			lowercase					:	List[Any]									=True
			lowercase					:	Optional[int]									=False
			lowercase					:	Union[str, Any]									=False
			def 		UpperCamelCase  (     self		):
								lowercase_       :Union[str, Any]     =       TFBlenderbotModelTester(self		)
								lowercase_       :Tuple     =       ConfigTester(self   ,							config_class=UpperCAmelCase_		)
			def 		UpperCamelCase  (     self		):
								self.config_tester.run_common_tests()




			def 		UpperCamelCase  (     self		):
								lowercase_       :List[Any]     =       self.model_tester.prepare_config_and_inputs_for_common()
								self.model_tester.check_decoder_model_past_large_inputs(*UpperCAmelCase_		)






@require_tokenizers
@require_tf
class      UpperCamelCase					(		unittest.TestCase     ):


			'''simple docstring'''







			lowercase					:	Dict									=["""My friends are cool but they eat too many carbs."""]
			lowercase					:	Union[str, Any]									="""facebook/blenderbot-400M-distill"""
			@cached_property
			def 		UpperCamelCase  (     self		):
								return BlenderbotTokenizer.from_pretrained(self.model_name		)
			@cached_property
			def 		UpperCamelCase  (     self		):
								lowercase_       :int     =       TFAutoModelForSeqaSeqLM.from_pretrained(self.model_name		)
								return model




			@slow
			def 		UpperCamelCase  (     self		):
								lowercase_       :Dict     =       self.tokenizer(self.src_text   ,							return_tensors='''tf'''		)
								lowercase_       :List[Any]     =       self.model.generate(
								    model_inputs.input_ids   ,							)
								lowercase_       :Any     =       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?"
								)





from io import BytesIO
from typing import List, Union

import requests

from ..utils import add_end_docstrings, is_decord_available, is_torch_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_decord_available():
   import numpy as np
   from decord import VideoReader


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

SCREAMING_SNAKE_CASE       :		List[str]         =				logging.get_logger(__name__)






@add_end_docstrings(lowercase__     )
class      UpperCamelCase					(		lowercase__     ):


   '''simple docstring'''







   def __init__(     self   ,							*UpperCamelCase_   ,							**UpperCamelCase_		):
        super().__init__(*UpperCamelCase_   ,							**UpperCamelCase_		)
        requires_backends(self   ,							'''decord'''		)
        self.check_model_type(UpperCamelCase_		)
   def 		UpperCamelCase  (     self   ,							UpperCamelCase_=None   ,							UpperCamelCase_=None   ,							UpperCamelCase_=None		):
        lowercase_       :int     =       {}
        if frame_sampling_rate is not None:
             lowercase_       :int     =       frame_sampling_rate
        if num_frames is not None:
             lowercase_       :int     =       num_frames

        lowercase_       :str     =       {}
        if top_k is not None:
             lowercase_       :Optional[int]     =       top_k
        return preprocess_params, {}, postprocess_params
   def __call__(     self   ,							UpperCamelCase_   ,							**UpperCamelCase_		):
        return super().__call__(UpperCamelCase_   ,							**UpperCamelCase_		)
   def 		UpperCamelCase  (     self   ,							UpperCamelCase_   ,							UpperCamelCase_=None   ,							UpperCamelCase_=1		):
        if num_frames is None:
             lowercase_       :str     =       self.model.config.num_frames

        if video.startswith('''http://'''		) or video.startswith('''https://'''		):
             lowercase_       :str     =       BytesIO(requests.get(UpperCamelCase_		).content		)

        lowercase_       :Optional[int]     =       VideoReader(UpperCamelCase_		)
        videoreader.seek(0		)

        lowercase_       :Tuple     =       0
        lowercase_       :Optional[Any]     =       num_frames * frame_sampling_rate - 1
        lowercase_       :Any     =       np.linspace(UpperCamelCase_   ,							UpperCamelCase_   ,							num=UpperCamelCase_   ,							dtype=np.intaa		)

        lowercase_       :Dict     =       videoreader.get_batch(UpperCamelCase_		).asnumpy()
        lowercase_       :List[Any]     =       list(UpperCamelCase_		)

        lowercase_       :Any     =       self.image_processor(UpperCamelCase_   ,							return_tensors=self.framework		)
        return model_inputs
   def 		UpperCamelCase  (     self   ,							UpperCamelCase_		):
        lowercase_       :List[str]     =       self.model(**UpperCamelCase_		)
        return model_outputs




   def 		UpperCamelCase  (     self   ,							UpperCamelCase_   ,							UpperCamelCase_=5		):
        if top_k > self.model.config.num_labels:
             lowercase_       :List[str]     =       self.model.config.num_labels

        if self.framework == "pt":
             lowercase_       :Optional[int]     =       model_outputs.logits.softmax(-1		)[0]
             lowercase_    ,      lowercase_       :Dict     =       probs.topk(UpperCamelCase_		)
        else:
             raise ValueError(f"Unsupported framework: {self.framework}"		)

        lowercase_       :Dict     =       scores.tolist()
        lowercase_       :Any     =       ids.tolist()
        return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(UpperCamelCase_   ,							UpperCamelCase_		)]

