
import os
import unicodedata
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


lowerCamelCase	   =logging.get_logger(__name__)
lowerCamelCase	   ={"vocab_file": "spiece.model"}

lowerCamelCase	   ={
    "vocab_file": {
        "albert-base-v1": "https://huggingface.co/albert-base-v1/resolve/main/spiece.model",
        "albert-large-v1": "https://huggingface.co/albert-large-v1/resolve/main/spiece.model",
        "albert-xlarge-v1": "https://huggingface.co/albert-xlarge-v1/resolve/main/spiece.model",
        "albert-xxlarge-v1": "https://huggingface.co/albert-xxlarge-v1/resolve/main/spiece.model",
        "albert-base-v2": "https://huggingface.co/albert-base-v2/resolve/main/spiece.model",
        "albert-large-v2": "https://huggingface.co/albert-large-v2/resolve/main/spiece.model",
        "albert-xlarge-v2": "https://huggingface.co/albert-xlarge-v2/resolve/main/spiece.model",
        "albert-xxlarge-v2": "https://huggingface.co/albert-xxlarge-v2/resolve/main/spiece.model",
    }
}

lowerCamelCase	   ={
    "albert-base-v1": 5_1_2,
    "albert-large-v1": 5_1_2,
    "albert-xlarge-v1": 5_1_2,
    "albert-xxlarge-v1": 5_1_2,
    "albert-base-v2": 5_1_2,
    "albert-large-v2": 5_1_2,
    "albert-xlarge-v2": 5_1_2,
    "albert-xxlarge-v2": 5_1_2,
}

lowerCamelCase	   ="▁"




class 				_lowerCamelCase (      UpperCamelCase_ ):





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		   =  VOCAB_FILES_NAMES
     SCREAMING_SNAKE_CASE_		   =  PRETRAINED_VOCAB_FILES_MAP
     SCREAMING_SNAKE_CASE_		   =  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES






     def __init__(							self						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE=True						,						__SCREAMING_SNAKE_CASE=True						,						__SCREAMING_SNAKE_CASE=False						,						__SCREAMING_SNAKE_CASE="[CLS]"						,						__SCREAMING_SNAKE_CASE="[SEP]"						,						__SCREAMING_SNAKE_CASE="<unk>"						,						__SCREAMING_SNAKE_CASE="[SEP]"						,						__SCREAMING_SNAKE_CASE="<pad>"						,						__SCREAMING_SNAKE_CASE="[CLS]"						,						__SCREAMING_SNAKE_CASE="[MASK]"						,						__SCREAMING_SNAKE_CASE = None						,						**__SCREAMING_SNAKE_CASE						,						)  ->   None:




           """simple docstring"""







           UpperCamelCase__    :   Union[str, Any]					=				(
               AddedToken(__SCREAMING_SNAKE_CASE						,						lstrip=__SCREAMING_SNAKE_CASE						,						rstrip=__SCREAMING_SNAKE_CASE						,						normalized=__SCREAMING_SNAKE_CASE  )
               if isinstance(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE  )
               else mask_token
           )

           UpperCamelCase__    :   List[Any]					=				{} if sp_model_kwargs is None else sp_model_kwargs

           super().__init__(
               do_lower_case=__SCREAMING_SNAKE_CASE						,						remove_space=__SCREAMING_SNAKE_CASE						,						keep_accents=__SCREAMING_SNAKE_CASE						,						bos_token=__SCREAMING_SNAKE_CASE						,						eos_token=__SCREAMING_SNAKE_CASE						,						unk_token=__SCREAMING_SNAKE_CASE						,						sep_token=__SCREAMING_SNAKE_CASE						,						pad_token=__SCREAMING_SNAKE_CASE						,						cls_token=__SCREAMING_SNAKE_CASE						,						mask_token=__SCREAMING_SNAKE_CASE						,						sp_model_kwargs=self.sp_model_kwargs						,						**__SCREAMING_SNAKE_CASE						,						)

           UpperCamelCase__    :   int					=				do_lower_case
           UpperCamelCase__    :   int					=				remove_space
           UpperCamelCase__    :   Tuple					=				keep_accents
           UpperCamelCase__    :   List[Any]					=				vocab_file

           UpperCamelCase__    :   Optional[int]					=				spm.SentencePieceProcessor(**self.sp_model_kwargs  )
           self.sp_model.Load(__SCREAMING_SNAKE_CASE  )






     @property
     def    __SCREAMING_SNAKE_CASE			(							self  )  ->   str:




           """simple docstring"""







           return len(self.sp_model  )






     def    __SCREAMING_SNAKE_CASE			(							self  )  ->   str:




           """simple docstring"""







           UpperCamelCase__    :   Optional[Any]					=				{self.convert_ids_to_tokens(__SCREAMING_SNAKE_CASE  ): i for i in range(self.vocab_size  )}
           vocab.update(self.added_tokens_encoder  )
           return vocab






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




           """simple docstring"""







           UpperCamelCase__    :   Any					=				self.__dict__.copy()
           UpperCamelCase__    :   Union[str, Any]					=				None
           return state






     def __setstate__(							self						,						__SCREAMING_SNAKE_CASE  )  ->   str:




           """simple docstring"""







           UpperCamelCase__    :   int					=				d

           # for backward compatibility
           if not hasattr(self						,						'''sp_model_kwargs'''  ):
                 UpperCamelCase__    :   List[Any]					=				{}

           UpperCamelCase__    :   Optional[Any]					=				spm.SentencePieceProcessor(**self.sp_model_kwargs  )
           self.sp_model.Load(self.vocab_file  )






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




           """simple docstring"""







           if self.remove_space:
                 UpperCamelCase__    :   int					=				''' '''.join(inputs.strip().split()  )
           else:
                 UpperCamelCase__    :   Optional[int]					=				inputs
           UpperCamelCase__    :   int					=				outputs.replace('''``'''						,						'''"'''  ).replace('''\'\''''						,						'''"'''  )

           if not self.keep_accents:
                 UpperCamelCase__    :   str					=				unicodedata.normalize('''NFKD'''						,						__SCREAMING_SNAKE_CASE  )
                 UpperCamelCase__    :   Optional[Any]					=				''''''.join([c for c in outputs if not unicodedata.combining(__SCREAMING_SNAKE_CASE  )]  )
           if self.do_lower_case:
                 UpperCamelCase__    :   Union[str, Any]					=				outputs.lower()

           return outputs






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




           """simple docstring"""







           UpperCamelCase__    :   Union[str, Any]					=				self.preprocess_text(__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   Union[str, Any]					=				self.sp_model.encode(__SCREAMING_SNAKE_CASE						,						out_type=__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   List[str]					=				[]
           for piece in pieces:
                 if len(__SCREAMING_SNAKE_CASE  ) > 1 and piece[-1] == str(''','''  ) and piece[-2].isdigit():
                       UpperCamelCase__    :   Tuple					=				self.sp_model.EncodeAsPieces(piece[:-1].replace(__SCREAMING_SNAKE_CASE						,						''''''  )  )
                       if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
                             if len(cur_pieces[0]  ) == 1:
                                   UpperCamelCase__    :   Tuple					=				cur_pieces[1:]
                             else:
                                   UpperCamelCase__    :   List[Any]					=				cur_pieces[0][1:]
                       cur_pieces.append(piece[-1]  )
                       new_pieces.extend(__SCREAMING_SNAKE_CASE  )
                 else:
                       new_pieces.append(__SCREAMING_SNAKE_CASE  )

           return new_pieces






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE  )  ->   Tuple:




           """simple docstring"""







           return self.sp_model.PieceToId(__SCREAMING_SNAKE_CASE  )






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




           """simple docstring"""







           return self.sp_model.IdToPiece(__SCREAMING_SNAKE_CASE  )






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE  )  ->   Any:




           """simple docstring"""







           UpperCamelCase__    :   Tuple					=				[]
           UpperCamelCase__    :   List[Any]					=				''''''
           UpperCamelCase__    :   str					=				False
           for token in tokens:
                 # make sure that special tokens are not decoded using sentencepiece model
                 if token in self.all_special_tokens:
                       if not prev_is_special:
                             out_string += " "
                       out_string += self.sp_model.decode(__SCREAMING_SNAKE_CASE  ) + token
                       UpperCamelCase__    :   str					=				True
                       UpperCamelCase__    :   str					=				[]
                 else:
                       current_sub_tokens.append(__SCREAMING_SNAKE_CASE  )
                       UpperCamelCase__    :   Optional[Any]					=				False
           out_string += self.sp_model.decode(__SCREAMING_SNAKE_CASE  )
           return out_string.strip()






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE = None  )  ->   List[int]:




           """simple docstring"""







           UpperCamelCase__    :   Optional[Any]					=				[self.sep_token_id]
           UpperCamelCase__    :   Tuple					=				[self.cls_token_id]
           if token_ids_a is None:
                 return cls + token_ids_a + sep
           return cls + token_ids_a + sep + token_ids_a + sep






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE = None						,						__SCREAMING_SNAKE_CASE = False  )  ->   List[int]:




           """simple docstring"""







           if already_has_special_tokens:
                 return super().get_special_tokens_mask(
                     token_ids_a=__SCREAMING_SNAKE_CASE						,						token_ids_a=__SCREAMING_SNAKE_CASE						,						already_has_special_tokens=__SCREAMING_SNAKE_CASE  )

           if token_ids_a is not None:
                 return [1] + ([0] * len(__SCREAMING_SNAKE_CASE  )) + [1] + ([0] * len(__SCREAMING_SNAKE_CASE  )) + [1]
           return [1] + ([0] * len(__SCREAMING_SNAKE_CASE  )) + [1]






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE = None  )  ->   List[int]:




           """simple docstring"""







           UpperCamelCase__    :   List[str]					=				[self.sep_token_id]
           UpperCamelCase__    :   int					=				[self.cls_token_id]

           if token_ids_a is None:
                 return len(cls + token_ids_a + sep  ) * [0]
           return len(cls + token_ids_a + sep  ) * [0] + len(token_ids_a + sep  ) * [1]






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE = None  )  ->   Tuple[str]:




           """simple docstring"""







           if not os.path.isdir(__SCREAMING_SNAKE_CASE  ):
                 logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''  )
                 return
           UpperCamelCase__    :   List[str]					=				os.path.join(
               __SCREAMING_SNAKE_CASE						,						(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']  )

           if os.path.abspath(self.vocab_file  ) != os.path.abspath(__SCREAMING_SNAKE_CASE  ) and os.path.isfile(self.vocab_file  ):
                 copyfile(self.vocab_file						,						__SCREAMING_SNAKE_CASE  )
           elif not os.path.isfile(self.vocab_file  ):
                 with open(__SCREAMING_SNAKE_CASE						,						'''wb'''  ) as fi:
                       UpperCamelCase__    :   Dict					=				self.sp_model.serialized_model_proto()
                       fi.write(__SCREAMING_SNAKE_CASE  )

           return (out_vocab_file,)




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  )]


