
'''simple docstring'''




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

lowercase     :							Tuple       						=     logging.get_logger(__name__)





@add_end_docstrings(__snake_case		)
class 					A							(				__snake_case		):





     def __init__(				self		,     *SCREAMING_SNAKE_CASE		,     **SCREAMING_SNAKE_CASE      )					->							Any:






         """simple docstring"""






         super().__init__(*SCREAMING_SNAKE_CASE		,     **SCREAMING_SNAKE_CASE      )
         requires_backends(self		,     '''decord'''      )
         self.check_model_type(SCREAMING_SNAKE_CASE      )





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE=None		,     SCREAMING_SNAKE_CASE=None		,     SCREAMING_SNAKE_CASE=None      )					->							Optional[Any]:






         """simple docstring"""






         A			:  Tuple   =		{}
         if frame_sampling_rate is not None:
             A			:  Optional[int]   =		frame_sampling_rate
         if num_frames is not None:
             A			:  Optional[Any]   =		num_frames

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





     def __call__(				self		,     SCREAMING_SNAKE_CASE		,     **SCREAMING_SNAKE_CASE      )					->							Dict:






         """simple docstring"""






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





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE=None		,     SCREAMING_SNAKE_CASE=1      )					->							int:






         """simple docstring"""






         if num_frames is None:
             A			:  List[Any]   =		self.model.config.num_frames

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

         A			:  List[Any]   =		VideoReader(SCREAMING_SNAKE_CASE      )
         videoreader.seek(0      )

         A			:  List[Any]   =		0
         A			:  str   =		num_frames * frame_sampling_rate - 1
         A			:  Tuple   =		np.linspace(SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE		,     num=SCREAMING_SNAKE_CASE		,     dtype=np.intaa      )

         A			:  Optional[int]   =		videoreader.get_batch(SCREAMING_SNAKE_CASE      ).asnumpy()
         A			:  Union[str, Any]   =		list(SCREAMING_SNAKE_CASE      )

         A			:  Dict   =		self.image_processor(SCREAMING_SNAKE_CASE		,     return_tensors=self.framework      )
         return model_inputs





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE      )					->							Union[str, Any]:






         """simple docstring"""






         A			:  Dict   =		self.model(**SCREAMING_SNAKE_CASE      )
         return model_outputs





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE=5      )					->							Any:






         """simple docstring"""






         if top_k > self.model.config.num_labels:
             A			:  Dict   =		self.model.config.num_labels

         if self.framework == "pt":
             A			:  Optional[Any]   =		model_outputs.logits.softmax(-1      )[0]
             A,    A			:  Any   =		probs.topk(SCREAMING_SNAKE_CASE      )
         else:
             raise ValueError(F'Unsupported framework: {self.framework}'      )

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






'''simple docstring'''




import json
import os
import unittest

from transformers import DebertaTokenizer, DebertaTokenizerFast
from transformers.models.deberta.tokenization_deberta import VOCAB_FILES_NAMES
from transformers.testing_utils import slow

from ...test_tokenization_common import TokenizerTesterMixin





class 					A							(				__snake_case       ,     unittest.TestCase		):
     __magic_name__          =						DebertaTokenizer
     __magic_name__          =						True
     __magic_name__          =						DebertaTokenizerFast





     def 							__lowerCAmelCase				(				self      )					->							Tuple:






         """simple docstring"""






         super().setUp()

         # Adapted from Sennrich et al. 2015 and https://github.com/rsennrich/subword-nmt
         A			:  Optional[int]   =		[
             '''l''',
             '''o''',
             '''w''',
             '''e''',
             '''r''',
             '''s''',
             '''t''',
             '''i''',
             '''d''',
             '''n''',
             '''\u0120''',
             '''\u0120l''',
             '''\u0120n''',
             '''\u0120lo''',
             '''\u0120low''',
             '''er''',
             '''\u0120lowest''',
             '''\u0120newer''',
             '''\u0120wider''',
             '''[UNK]''',
         ]
         A			:  List[str]   =		dict(zip(SCREAMING_SNAKE_CASE		,     range(len(SCREAMING_SNAKE_CASE      )      )      )      )
         A			:  List[Any]   =		['''#version: 0.2''', '''\u0120 l''', '''\u0120l o''', '''\u0120lo w''', '''e r''', '''''']
         A			:  Tuple   =		{'''unk_token''': '''[UNK]'''}

         A			:  Optional[int]   =		os.path.join(self.tmpdirname		,     VOCAB_FILES_NAMES['''vocab_file''']      )
         A			:  str   =		os.path.join(self.tmpdirname		,     VOCAB_FILES_NAMES['''merges_file''']      )
         with open(self.vocab_file		,     '''w'''		,     encoding='''utf-8'''      ) as fp:
             fp.write(json.dumps(SCREAMING_SNAKE_CASE      ) + '''\n'''      )
         with open(self.merges_file		,     '''w'''		,     encoding='''utf-8'''      ) as fp:
             fp.write('''\n'''.join(SCREAMING_SNAKE_CASE      )      )





     def 							__lowerCAmelCase				(				self		,     **SCREAMING_SNAKE_CASE      )					->							Optional[Any]:






         """simple docstring"""






         kwargs.update(self.special_tokens_map      )
         return self.tokenizer_class.from_pretrained(self.tmpdirname		,     **SCREAMING_SNAKE_CASE      )





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE      )					->							str:






         """simple docstring"""






         A			:  Union[str, Any]   =		'''lower newer'''
         A			:  List[Any]   =		'''lower newer'''
         return input_text, output_text





     def 							__lowerCAmelCase				(				self      )					->							str:






         """simple docstring"""






         A			:  str   =		self.get_tokenizer()
         A			:  List[str]   =		'''lower newer'''
         A			:  int   =		['''l''', '''o''', '''w''', '''er''', '''\u0120''', '''n''', '''e''', '''w''', '''er''']
         A			:  Any   =		tokenizer.tokenize(SCREAMING_SNAKE_CASE      )
         self.assertListEqual(SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE      )

         A			:  str   =		tokens + [tokenizer.unk_token]
         A			:  str   =		[0, 1, 2, 15, 10, 9, 3, 2, 15, 19]
         self.assertListEqual(tokenizer.convert_tokens_to_ids(SCREAMING_SNAKE_CASE      )		,     SCREAMING_SNAKE_CASE      )





     def 							__lowerCAmelCase				(				self      )					->							str:






         """simple docstring"""






         A			:  List[Any]   =		self.get_tokenizer()
         A			:  Tuple   =		tokenizer('''Hello'''		,     '''World'''      )
         A			:  str   =		[0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1]
         self.assertListEqual(tokd['''token_type_ids''']		,     SCREAMING_SNAKE_CASE      )





     @slow
     def 							__lowerCAmelCase				(				self      )					->							int:






         """simple docstring"""






         A			:  Union[str, Any]   =		self.tokenizer_class.from_pretrained('''microsoft/deberta-base'''      )

         A			:  Union[str, Any]   =		tokenizer.encode('''sequence builders'''		,     add_special_tokens=SCREAMING_SNAKE_CASE      )
         A			:  Any   =		tokenizer.encode('''multi-sequence build'''		,     add_special_tokens=SCREAMING_SNAKE_CASE      )

         A			:  Tuple   =		tokenizer.encode(
             '''sequence builders'''		,     add_special_tokens=SCREAMING_SNAKE_CASE		,     add_prefix_space=SCREAMING_SNAKE_CASE      )
         A			:  Union[str, Any]   =		tokenizer.encode(
             '''sequence builders'''		,     '''multi-sequence build'''		,     add_special_tokens=SCREAMING_SNAKE_CASE		,     add_prefix_space=SCREAMING_SNAKE_CASE      )

         A			:  List[str]   =		tokenizer.build_inputs_with_special_tokens(SCREAMING_SNAKE_CASE      )
         A			:  Union[str, Any]   =		tokenizer.build_inputs_with_special_tokens(SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE      )

         assert encoded_sentence == encoded_text_from_decode
         assert encoded_pair == encoded_pair_from_decode





     @slow
     def 							__lowerCAmelCase				(				self      )					->							int:






         """simple docstring"""






         A			:  Any   =		[self.tokenizer_class]
         if self.test_rust_tokenizer:
             tokenizer_classes.append(self.rust_tokenizer_class      )

         for tokenizer_class in tokenizer_classes:
             A			:  List[str]   =		tokenizer_class.from_pretrained('''microsoft/deberta-base'''      )

             A			:  Dict   =		[
                 '''ALBERT: A Lite BERT for Self-supervised Learning of Language Representations''',
                 '''ALBERT incorporates two parameter reduction techniques''',
                 '''The first one is a factorized embedding parameterization. By decomposing the large vocabulary'''
                 ''' embedding matrix into two small matrices, we separate the size of the hidden layers from the size of'''
                 ''' vocabulary embedding.''',
             ]

             A			:  List[Any]   =		tokenizer(SCREAMING_SNAKE_CASE		,     padding=SCREAMING_SNAKE_CASE      )
             A			:  List[str]   =		[tokenizer.decode(SCREAMING_SNAKE_CASE		,     skip_special_tokens=SCREAMING_SNAKE_CASE      ) for seq in encoding['''input_ids''']]

             # fmt: off
             A			:  List[str]   =		{
                 '''input_ids''': [
                     [1, 2118, 11126, 565, 35, 83, 25191, 163, 18854, 13, 12156, 12, 16101, 25376, 13807, 9, 22205, 27893, 1635, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                     [1, 2118, 11126, 565, 24536, 80, 43797, 4878, 7373, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                     [1, 133, 78, 65, 16, 10, 3724, 1538, 33183, 11303, 43797, 1938, 4, 870, 24165, 29105, 5, 739, 32644, 33183, 11303, 36173, 88, 80, 650, 7821, 45940, 6, 52, 2559, 5, 1836, 9, 5, 7397, 13171, 31, 5, 1836, 9, 32644, 33183, 11303, 4, 2]
                 ],
                 '''token_type_ids''': [
                     [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                     [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                     [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
                 ],
                 '''attention_mask''': [
                     [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                     [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
                     [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]
                 ]
             }
             # fmt: on

             A			:  Any   =		[
                 '''ALBERT: A Lite BERT for Self-supervised Learning of Language Representations''',
                 '''ALBERT incorporates two parameter reduction techniques''',
                 '''The first one is a factorized embedding parameterization. By decomposing the large vocabulary'''
                 ''' embedding matrix into two small matrices, we separate the size of the hidden layers from the size of'''
                 ''' vocabulary embedding.''',
             ]

             self.assertDictEqual(encoding.data		,     SCREAMING_SNAKE_CASE      )

             for expected, decoded in zip(SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE      ):
                 self.assertEqual(SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE      )




