
from __future__ import annotations

import unittest

from transformers import DistilBertConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
       import tensorflow as tf

       from transformers.models.distilbert.modeling_tf_distilbert import (
           TF_DISTILBERT_PRETRAINED_MODEL_ARCHIVE_LIST,
           TFDistilBertForMaskedLM,
           TFDistilBertForMultipleChoice,
           TFDistilBertForQuestionAnswering,
           TFDistilBertForSequenceClassification,
           TFDistilBertForTokenClassification,
           TFDistilBertModel,
       )




class 						__lowercase		:




      """simple docstring"""





      def __init__(				self    ,				A    ,				)		->    Union[str, Any]:
         snake_case				:    List[Any]           =	parent
         snake_case				:    str           =	1_3
         snake_case				:    Tuple           =	7
         snake_case				:    Tuple           =	True
         snake_case				:    Union[str, Any]           =	True
         snake_case				:    Optional[Any]           =	False
         snake_case				:    Union[str, Any]           =	True
         snake_case				:    Union[str, Any]           =	9_9
         snake_case				:    Any           =	3_2
         snake_case				:    Optional[Any]           =	2
         snake_case				:    Union[str, Any]           =	4
         snake_case				:    int           =	3_7
         snake_case				:    Any           =	"""gelu"""
         snake_case				:    Tuple           =	0.1
         snake_case				:    Optional[Any]           =	0.1
         snake_case				:    List[str]           =	5_1_2
         snake_case				:    Tuple           =	1_6
         snake_case				:    Optional[Any]           =	2
         snake_case				:    Any           =	0.02
         snake_case				:    List[str]           =	3
         snake_case				:    Dict           =	4
         snake_case				:    Dict           =	None

      def      UpperCAmelCase				(				self     )		->    Union[str, Any]:
         snake_case				:    List[str]           =	ids_tensor([self.batch_size, self.seq_length]    ,				self.vocab_size     )

         snake_case				:    Dict           =	None
         if self.use_input_mask:
            snake_case				:    List[Any]           =	random_attention_mask([self.batch_size, self.seq_length]     )

         snake_case				:    Tuple           =	None
         snake_case				:    Optional[Any]           =	None
         snake_case				:    Any           =	None
         if self.use_labels:
            snake_case				:    List[str]           =	ids_tensor([self.batch_size]    ,				self.type_sequence_label_size     )
            snake_case				:    List[Any]           =	ids_tensor([self.batch_size, self.seq_length]    ,				self.num_labels     )
            snake_case				:    str           =	ids_tensor([self.batch_size]    ,				self.num_choices     )

         snake_case				:    List[str]           =	DistilBertConfig(
             vocab_size=self.vocab_size    ,				dim=self.hidden_size    ,				n_layers=self.num_hidden_layers    ,				n_heads=self.num_attention_heads    ,				hidden_dim=self.intermediate_size    ,				hidden_act=self.hidden_act    ,				dropout=self.hidden_dropout_prob    ,				attention_dropout=self.attention_probs_dropout_prob    ,				max_position_embeddings=self.max_position_embeddings    ,				initializer_range=self.initializer_range    ,				)

         return config, input_ids, input_mask, sequence_labels, token_labels, choice_labels

      def      UpperCAmelCase				(				self    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A     )		->    List[Any]:
         snake_case				:    List[str]           =	TFDistilBertModel(config=_SCREAMING_SNAKE_CASE     )
         snake_case				:    List[Any]           =	{"""input_ids""": input_ids, """attention_mask""": input_mask}

         snake_case				:    List[Any]           =	model(_SCREAMING_SNAKE_CASE     )

         snake_case				:    List[str]           =	[input_ids, input_mask]

         snake_case				:    int           =	model(_SCREAMING_SNAKE_CASE     )

         self.parent.assertEqual(result.last_hidden_state.shape    ,				(self.batch_size, self.seq_length, self.hidden_size)     )

      def      UpperCAmelCase				(				self    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A     )		->    Optional[int]:
         snake_case				:    List[str]           =	TFDistilBertForMaskedLM(config=_SCREAMING_SNAKE_CASE     )
         snake_case				:    Optional[Any]           =	{"""input_ids""": input_ids, """attention_mask""": input_mask}
         snake_case				:    Union[str, Any]           =	model(_SCREAMING_SNAKE_CASE     )
         self.parent.assertEqual(result.logits.shape    ,				(self.batch_size, self.seq_length, self.vocab_size)     )

      def      UpperCAmelCase				(				self    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A     )		->    int:
         snake_case				:    Union[str, Any]           =	TFDistilBertForQuestionAnswering(config=_SCREAMING_SNAKE_CASE     )
         snake_case				:    Optional[int]           =	{
             """input_ids""": input_ids,
             """attention_mask""": input_mask,
         }
         snake_case				:    Optional[int]           =	model(_SCREAMING_SNAKE_CASE     )
         self.parent.assertEqual(result.start_logits.shape    ,				(self.batch_size, self.seq_length)     )
         self.parent.assertEqual(result.end_logits.shape    ,				(self.batch_size, self.seq_length)     )

      def      UpperCAmelCase				(				self    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A     )		->    Optional[int]:
         snake_case				:    Union[str, Any]           =	self.num_labels
         snake_case				:    Optional[int]           =	TFDistilBertForSequenceClassification(_SCREAMING_SNAKE_CASE     )
         snake_case				:    List[str]           =	{"""input_ids""": input_ids, """attention_mask""": input_mask}
         snake_case				:    Optional[int]           =	model(_SCREAMING_SNAKE_CASE     )
         self.parent.assertEqual(result.logits.shape    ,				(self.batch_size, self.num_labels)     )

      def      UpperCAmelCase				(				self    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A     )		->    Optional[int]:
         snake_case				:    List[str]           =	self.num_choices
         snake_case				:    Tuple           =	TFDistilBertForMultipleChoice(_SCREAMING_SNAKE_CASE     )
         snake_case				:    List[str]           =	tf.tile(tf.expand_dims(_SCREAMING_SNAKE_CASE    ,				1     )    ,				(1, self.num_choices, 1)     )
         snake_case				:    Union[str, Any]           =	tf.tile(tf.expand_dims(_SCREAMING_SNAKE_CASE    ,				1     )    ,				(1, self.num_choices, 1)     )
         snake_case				:    int           =	{
             """input_ids""": multiple_choice_inputs_ids,
             """attention_mask""": multiple_choice_input_mask,
         }
         snake_case				:    Optional[int]           =	model(_SCREAMING_SNAKE_CASE     )
         self.parent.assertEqual(result.logits.shape    ,				(self.batch_size, self.num_choices)     )

      def      UpperCAmelCase				(				self    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A     )		->    Tuple:
         snake_case				:    Tuple           =	self.num_labels
         snake_case				:    Any           =	TFDistilBertForTokenClassification(_SCREAMING_SNAKE_CASE     )
         snake_case				:    Any           =	{"""input_ids""": input_ids, """attention_mask""": input_mask}
         snake_case				:    List[Any]           =	model(_SCREAMING_SNAKE_CASE     )
         self.parent.assertEqual(result.logits.shape    ,				(self.batch_size, self.seq_length, self.num_labels)     )

      def      UpperCAmelCase				(				self     )		->    Any:
         snake_case				:    Tuple           =	self.prepare_config_and_inputs()
         (snake_case)				:    Dict           =	config_and_inputs
         snake_case				:    Any           =	{"""input_ids""": input_ids, """attention_mask""": input_mask}
         return config, inputs_dict




@require_tf
class 						__lowercase		(UpperCAmelCase__					,		UpperCAmelCase__					,		unittest.TestCase							):




      """simple docstring"""





      _snake_case												=						(
          (
              TFDistilBertModel,
              TFDistilBertForMaskedLM,
              TFDistilBertForQuestionAnswering,
              TFDistilBertForSequenceClassification,
              TFDistilBertForTokenClassification,
              TFDistilBertForMultipleChoice,
          )
          if is_tf_available()
          else None
      )
      _snake_case												=						(
          {
              'feature-extraction': TFDistilBertModel,
              'fill-mask': TFDistilBertForMaskedLM,
              'question-answering': TFDistilBertForQuestionAnswering,
              'text-classification': TFDistilBertForSequenceClassification,
              'token-classification': TFDistilBertForTokenClassification,
              'zero-shot': TFDistilBertForSequenceClassification,
          }
          if is_tf_available()
          else {}
      )
      _snake_case												=						False
      _snake_case												=						False

      def      UpperCAmelCase				(				self     )		->    Optional[int]:
         snake_case				:    Tuple           =	TFDistilBertModelTester(self     )
         snake_case				:    Dict           =	ConfigTester(self    ,				config_class=_SCREAMING_SNAKE_CASE    ,				dim=3_7     )

      def      UpperCAmelCase				(				self     )		->    Dict:
         self.config_tester.run_common_tests()

      def      UpperCAmelCase				(				self     )		->    Optional[int]:
         snake_case				:    str           =	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_distilbert_model(*_SCREAMING_SNAKE_CASE     )

      def      UpperCAmelCase				(				self     )		->    Optional[Any]:
         snake_case				:    str           =	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_distilbert_for_masked_lm(*_SCREAMING_SNAKE_CASE     )

      def      UpperCAmelCase				(				self     )		->    List[str]:
         snake_case				:    str           =	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_distilbert_for_question_answering(*_SCREAMING_SNAKE_CASE     )

      def      UpperCAmelCase				(				self     )		->    Optional[Any]:
         snake_case				:    List[Any]           =	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_distilbert_for_sequence_classification(*_SCREAMING_SNAKE_CASE     )

      def      UpperCAmelCase				(				self     )		->    int:
         snake_case				:    str           =	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_distilbert_for_multiple_choice(*_SCREAMING_SNAKE_CASE     )

      def      UpperCAmelCase				(				self     )		->    int:
         snake_case				:    int           =	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_distilbert_for_token_classification(*_SCREAMING_SNAKE_CASE     )

      @slow
      def      UpperCAmelCase				(				self     )		->    Optional[Any]:




         for model_name in list(TF_DISTILBERT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]     ):
            snake_case				:    Tuple           =	TFDistilBertModel.from_pretrained(_SCREAMING_SNAKE_CASE     )
            self.assertIsNotNone(_SCREAMING_SNAKE_CASE     )




@require_tf
class 						__lowercase		(unittest.TestCase							):




      """simple docstring"""





      @slow
      def      UpperCAmelCase				(				self     )		->    str:
         snake_case				:    str           =	TFDistilBertModel.from_pretrained("""distilbert-base-uncased"""     )
         snake_case				:    Union[str, Any]           =	tf.constant([[0, 1, 2, 3, 4, 5]]     )
         snake_case				:    int           =	model(_SCREAMING_SNAKE_CASE     )[0]

         snake_case				:    List[Any]           =	[1, 6, 7_6_8]
         self.assertEqual(output.shape    ,				_SCREAMING_SNAKE_CASE     )

         snake_case				:    List[Any]           =	tf.constant(
             [
                 [
                     [0.19_26_18_85, -0.13_73_29_55, 0.4_11_97_99],
                     [0.22_15_01_56, -0.07_42_26_61, 0.39_03_72_04],
                     [0.22_75_60_18, -0.0_89_64_14, 0.3_70_14_67],
                 ]
             ]     )
         tf.debugging.assert_near(output[:, :3, :3]    ,				_SCREAMING_SNAKE_CASE    ,				atol=1e-4     )











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

lowerCamelCase	:	int				=   logging.get_logger(__name__)




@add_end_docstrings(UpperCamelCase__							)
class 						__lowercase		(UpperCamelCase__							):




      """simple docstring"""





      def __init__(				self    ,				*A    ,				**A     )		->    Optional[int]:
         super().__init__(*A    ,				**A     )
         requires_backends(self    ,				"""decord"""     )
         self.check_model_type(A     )

      def      UpperCAmelCase				(				self    ,				A=None    ,				A=None    ,				A=None     )		->    int:
         snake_case				:    Any           =	{}
         if frame_sampling_rate is not None:
            snake_case				:    int           =	frame_sampling_rate
         if num_frames is not None:
            snake_case				:    Union[str, Any]           =	num_frames

         snake_case				:    str           =	{}
         if top_k is not None:
            snake_case				:    Optional[int]           =	top_k
         return preprocess_params, {}, postprocess_params

      def __call__(				self    ,				A    ,				**A     )		->    Dict:
         return super().__call__(A    ,				**A     )

      def      UpperCAmelCase				(				self    ,				A    ,				A=None    ,				A=1     )		->    Tuple:
         if num_frames is None:
            snake_case				:    Tuple           =	self.model.config.num_frames

         if video.startswith("""http://"""     ) or video.startswith("""https://"""     ):
            snake_case				:    Optional[int]           =	BytesIO(requests.get(A     ).content     )

         snake_case				:    Optional[Any]           =	VideoReader(A     )
         videoreader.seek(0     )

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

         snake_case				:    int           =	videoreader.get_batch(A     ).asnumpy()
         snake_case				:    List[Any]           =	list(A     )

         snake_case				:    Optional[Any]           =	self.image_processor(A    ,				return_tensors=self.framework     )
         return model_inputs

      def      UpperCAmelCase				(				self    ,				A     )		->    List[str]:
         snake_case				:    Dict           =	self.model(**A     )
         return model_outputs

      def      UpperCAmelCase				(				self    ,				A    ,				A=5     )		->    int:
         if top_k > self.model.config.num_labels:
            snake_case				:    str           =	self.model.config.num_labels

         if self.framework == "pt":
            snake_case				:    List[Any]           =	model_outputs.logits.softmax(-1     )[0]
            snake_case     , snake_case				:    Tuple           =	probs.topk(A     )
         else:
            raise ValueError(f"""Unsupported framework: {self.framework}"""     )

         snake_case				:    List[Any]           =	scores.tolist()
         snake_case				:    str           =	ids.tolist()
         return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(A    ,				A     )]





