
import json
import re
from typing import TYPE_CHECKING, List, Optional, Tuple, Union

import numpy as np

from ...utils import is_tf_available, is_torch_available, logging


if TYPE_CHECKING:
    if is_torch_available():
        import torch
    if is_tf_available():
        import tensorflow as tf

from tokenizers import pre_tokenizers

from ...tokenization_utils_base import BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from .tokenization_codegen import CodeGenTokenizer


lowerCamelCase	   =logging.get_logger(__name__)

lowerCamelCase	   ={"vocab_file": "vocab.json", "merges_file": "merges.txt", "tokenizer_file": "tokenizer.json"}

lowerCamelCase	   ={
    "vocab_file": {
        "Salesforce/codegen-350M-mono": "https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/vocab.json",
    },
    "merges_file": {
        "Salesforce/codegen-350M-mono": "https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/merges.txt",
    },
    "tokenizer_file": {
        "Salesforce/codegen-350M-mono": (
            "https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/tokenizer.json"
        ),
    },
}

lowerCamelCase	   ={
    "Salesforce/codegen-350M-mono": 2_0_4_8,
}




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
     SCREAMING_SNAKE_CASE_		   =  ['''input_ids''', '''attention_mask''']
     SCREAMING_SNAKE_CASE_		   =  CodeGenTokenizer






     def __init__(							self						,						__SCREAMING_SNAKE_CASE=None						,						__SCREAMING_SNAKE_CASE=None						,						__SCREAMING_SNAKE_CASE=None						,						__SCREAMING_SNAKE_CASE="<|endoftext|>"						,						__SCREAMING_SNAKE_CASE="<|endoftext|>"						,						__SCREAMING_SNAKE_CASE="<|endoftext|>"						,						__SCREAMING_SNAKE_CASE=False						,						**__SCREAMING_SNAKE_CASE						,						)  ->   Any:




           """simple docstring"""







           super().__init__(
               __SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE						,						tokenizer_file=__SCREAMING_SNAKE_CASE						,						unk_token=__SCREAMING_SNAKE_CASE						,						bos_token=__SCREAMING_SNAKE_CASE						,						eos_token=__SCREAMING_SNAKE_CASE						,						add_prefix_space=__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE						,						)

           if kwargs.pop('''add_bos_token'''						,						__SCREAMING_SNAKE_CASE  ):
                 UpperCamelCase__    :   int					=				kwargs.pop('''name_or_path'''						,						''''''  )
                 raise ValueError(
                     '''Currenty GPT2\'s fast tokenizer does NOT support adding a BOS token.'''
                     '''Instead you should use GPT2\'s slow tokenizer class `CodeGenTokenizer` as follows: \n'''
                     F'''`CodeGenTokenizer.from_pretrained(\'{model_id}\')`\nor\n'''
                     F'''`AutoTokenizer.from_pretrained(\'{model_id}\', use_fast=False)`\n'''
                     '''This issue will be fixed soon, see: https://github.com/huggingface/tokenizers/pull/1005.'''
                     ''' so that the fast tokenizer works correctly.'''  )

           UpperCamelCase__    :   Union[str, Any]					=				json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()  )
           if pre_tok_state.get('''add_prefix_space'''						,						__SCREAMING_SNAKE_CASE  ) != add_prefix_space:
                 UpperCamelCase__    :   Tuple					=				getattr(__SCREAMING_SNAKE_CASE						,						pre_tok_state.pop('''type'''  )  )
                 UpperCamelCase__    :   Tuple					=				add_prefix_space
                 UpperCamelCase__    :   Union[str, Any]					=				pre_tok_class(**__SCREAMING_SNAKE_CASE  )

           UpperCamelCase__    :   Union[str, Any]					=				add_prefix_space






     def    __SCREAMING_SNAKE_CASE			(							self						,						*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )  ->   BatchEncoding:




           """simple docstring"""







           UpperCamelCase__    :   Dict					=				kwargs.get('''is_split_into_words'''						,						__SCREAMING_SNAKE_CASE  )
           assert self.add_prefix_space or not is_split_into_words, (
               F'''You need to instantiate {self.__class__.__name__} with add_prefix_space=True '''
               "to use it with pretokenized inputs."
           )

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






     def    __SCREAMING_SNAKE_CASE			(							self						,						*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )  ->   BatchEncoding:




           """simple docstring"""







           UpperCamelCase__    :   Dict					=				kwargs.get('''is_split_into_words'''						,						__SCREAMING_SNAKE_CASE  )

           assert self.add_prefix_space or not is_split_into_words, (
               F'''You need to instantiate {self.__class__.__name__} with add_prefix_space=True '''
               "to use it with pretokenized inputs."
           )

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






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




           """simple docstring"""







           UpperCamelCase__    :   Optional[int]					=				self._tokenizer.model.save(__SCREAMING_SNAKE_CASE						,						name=__SCREAMING_SNAKE_CASE  )
           return tuple(__SCREAMING_SNAKE_CASE  )






     def    __SCREAMING_SNAKE_CASE			(							self						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE = False						,						__SCREAMING_SNAKE_CASE = None						,						__SCREAMING_SNAKE_CASE = None						,						**__SCREAMING_SNAKE_CASE						,						)  ->   str:




           """simple docstring"""







           UpperCamelCase__    :   Any					=				super().decode(
               token_ids=__SCREAMING_SNAKE_CASE						,						skip_special_tokens=__SCREAMING_SNAKE_CASE						,						clean_up_tokenization_spaces=__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE						,						)

           if truncate_before_pattern is not None and len(__SCREAMING_SNAKE_CASE  ) > 0:
                 UpperCamelCase__    :   Optional[int]					=				self.truncate(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE  )

           return decoded_text






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




           """simple docstring"""







           def find_re(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE  ):
                 UpperCamelCase__    :   List[str]					=				pattern.search(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE  )
                 return m.start() if m else -1

           UpperCamelCase__    :   Any					=				[re.compile(__SCREAMING_SNAKE_CASE						,						re.MULTILINE  ) for pattern in truncate_before_pattern]

           UpperCamelCase__    :   Dict					=				list(re.finditer('''^print'''						,						__SCREAMING_SNAKE_CASE						,						re.MULTILINE  )  )

           if len(__SCREAMING_SNAKE_CASE  ) > 1:
                 UpperCamelCase__    :   Optional[Any]					=				completion[: prints[1].start()]

           UpperCamelCase__    :   str					=				list(re.finditer('''^def'''						,						__SCREAMING_SNAKE_CASE						,						re.MULTILINE  )  )

           if len(__SCREAMING_SNAKE_CASE  ) > 1:
                 UpperCamelCase__    :   Optional[int]					=				completion[: defs[1].start()]

           UpperCamelCase__    :   int					=				0

           UpperCamelCase__    :   Optional[int]					=				[
               pos for pos in [find_re(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE  ) for terminal in terminals] if pos != -1
           ]

           if len(__SCREAMING_SNAKE_CASE  ) > 0:
                 return completion[: min(__SCREAMING_SNAKE_CASE  )]
           else:
                 return completion




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


