
import os
from typing import Dict, List, Union

import tensorflow as tf
from keras_nlp.tokenizers import BytePairTokenizer
from tensorflow_text import pad_model_inputs

from .tokenization_gpta import GPTaTokenizer




class 				_lowerCamelCase (      tf.keras.layers.Layer ):





     """simple docstring"""







     def __init__(							self						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE = None						,						__SCREAMING_SNAKE_CASE = None  )  ->   Optional[Any]:




           """simple docstring"""







           super().__init__()
           UpperCamelCase__    :   int					=				pad_token_id
           UpperCamelCase__    :   Union[str, Any]					=				max_length
           UpperCamelCase__    :   Optional[Any]					=				vocab
           UpperCamelCase__    :   Tuple					=				merges
           UpperCamelCase__    :   Optional[Any]					=				BytePairTokenizer(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE						,						sequence_length=__SCREAMING_SNAKE_CASE  )






     @classmethod
     def    __SCREAMING_SNAKE_CASE			(							cls						,						__SCREAMING_SNAKE_CASE						,						*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )  ->   Dict:




           """simple docstring"""







           UpperCamelCase__    :   str					=				[''' '''.join(__SCREAMING_SNAKE_CASE  ) for m in tokenizer.bpe_ranks.keys()]
           UpperCamelCase__    :   List[Any]					=				tokenizer.get_vocab()
           return cls(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE						,						*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )






     @classmethod
     def    __SCREAMING_SNAKE_CASE			(							cls						,						__SCREAMING_SNAKE_CASE						,						*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )  ->   Dict:




           """simple docstring"""







           UpperCamelCase__    :   List[str]					=				GPTaTokenizer.from_pretrained(__SCREAMING_SNAKE_CASE						,						*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )
           return cls.from_tokenizer(__SCREAMING_SNAKE_CASE						,						*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )






     @classmethod
     def    __SCREAMING_SNAKE_CASE			(							cls						,						__SCREAMING_SNAKE_CASE  )  ->   List[str]:




           """simple docstring"""







           return cls(**__SCREAMING_SNAKE_CASE  )






     def    __SCREAMING_SNAKE_CASE			(							self  )  ->   Dict:




           """simple docstring"""







           return {
               "vocab": self.vocab,
               "merges": self.merges,
               "max_length": self.max_length,
               "pad_token_id": self.pad_token_id,
           }






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




           """simple docstring"""







           UpperCamelCase__    :   List[str]					=				self.tf_tokenizer(__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   Optional[int]					=				tf.ones_like(__SCREAMING_SNAKE_CASE  )

           if self.pad_token_id is not None:
                 # pad the tokens up to max length
                 UpperCamelCase__    :   Tuple					=				max_length if max_length is not None else self.max_length

                 if max_length is not None:
                       UpperCamelCase__				,UpperCamelCase__    :   str					=				pad_model_inputs(
                           __SCREAMING_SNAKE_CASE						,						max_seq_length=__SCREAMING_SNAKE_CASE						,						pad_value=self.pad_token_id  )

           return {"attention_mask": attention_mask, "input_ids": input_ids}




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      =logging.get_logger(__name__)




@add_end_docstrings(UpperCamelCase_ )
class 				_lowerCamelCase (      UpperCamelCase_ ):





     """simple docstring"""







     def __init__(							self						,						*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )  ->   Dict:




           """simple docstring"""







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






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




           """simple docstring"""







           UpperCamelCase__    :   List[str]					=				{}
           if frame_sampling_rate is not None:
                 UpperCamelCase__    :   Tuple					=				frame_sampling_rate
           if num_frames is not None:
                 UpperCamelCase__    :   str					=				num_frames

           UpperCamelCase__    :   List[str]					=				{}
           if top_k is not None:
                 UpperCamelCase__    :   List[Any]					=				top_k
           return preprocess_params, {}, postprocess_params






     def __call__(							self						,						__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )  ->   Tuple:




           """simple docstring"""







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






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE=None						,						__SCREAMING_SNAKE_CASE=1  )  ->   Optional[Any]:




           """simple docstring"""







           if num_frames is None:
                 UpperCamelCase__    :   Optional[Any]					=				self.model.config.num_frames

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

           UpperCamelCase__    :   Tuple					=				VideoReader(__SCREAMING_SNAKE_CASE  )
           videoreader.seek(0  )

           UpperCamelCase__    :   Optional[int]					=				0
           UpperCamelCase__    :   str					=				num_frames * frame_sampling_rate - 1
           UpperCamelCase__    :   List[Any]					=				np.linspace(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE						,						num=__SCREAMING_SNAKE_CASE						,						dtype=np.intaa  )

           UpperCamelCase__    :   str					=				videoreader.get_batch(__SCREAMING_SNAKE_CASE  ).asnumpy()
           UpperCamelCase__    :   Tuple					=				list(__SCREAMING_SNAKE_CASE  )

           UpperCamelCase__    :   Dict					=				self.image_processor(__SCREAMING_SNAKE_CASE						,						return_tensors=self.framework  )
           return model_inputs






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




           """simple docstring"""







           UpperCamelCase__    :   Tuple					=				self.model(**__SCREAMING_SNAKE_CASE  )
           return model_outputs






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE=5  )  ->   str:




           """simple docstring"""







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

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

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


