
import re
import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin




class 				_lowerCamelCase (      UpperCamelCase_ ):





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		   =  ['''image_processor''', '''tokenizer''']
     SCREAMING_SNAKE_CASE_		   =  '''AutoImageProcessor'''
     SCREAMING_SNAKE_CASE_		   =  '''AutoTokenizer'''






     def __init__(							self						,						__SCREAMING_SNAKE_CASE=None						,						__SCREAMING_SNAKE_CASE=None						,						**__SCREAMING_SNAKE_CASE  )  ->   Union[str, Any]:




           """simple docstring"""







           UpperCamelCase__    :   List[str]					=				None
           if "feature_extractor" in kwargs:
                 warnings.warn(
                     '''The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'''
                     ''' instead.'''						,						__SCREAMING_SNAKE_CASE						,						)
                 UpperCamelCase__    :   Optional[Any]					=				kwargs.pop('''feature_extractor'''  )

           UpperCamelCase__    :   List[str]					=				image_processor if image_processor is not None else feature_extractor
           if image_processor is None:
                 raise ValueError('''You need to specify an `image_processor`.'''  )
           if tokenizer is None:
                 raise ValueError('''You need to specify a `tokenizer`.'''  )

           super().__init__(__SCREAMING_SNAKE_CASE						,						__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   Union[str, Any]					=				self.image_processor
           UpperCamelCase__    :   Union[str, Any]					=				False






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




           """simple docstring"""







           if self._in_target_context_manager:
                 return self.current_processor(*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )

           UpperCamelCase__    :   List[str]					=				kwargs.pop('''images'''						,						__SCREAMING_SNAKE_CASE  )
           UpperCamelCase__    :   List[Any]					=				kwargs.pop('''text'''						,						__SCREAMING_SNAKE_CASE  )
           if len(__SCREAMING_SNAKE_CASE  ) > 0:
                 UpperCamelCase__    :   Tuple					=				args[0]
                 UpperCamelCase__    :   Any					=				args[1:]

           if images is None and text is None:
                 raise ValueError('''You need to specify either an `images` or `text` input to process.'''  )

           if images is not None:
                 UpperCamelCase__    :   Any					=				self.image_processor(__SCREAMING_SNAKE_CASE						,						*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )
           if text is not None:
                 UpperCamelCase__    :   Optional[int]					=				self.tokenizer(__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )

           if text is None:
                 return inputs
           elif images is None:
                 return encodings
           else:
                 UpperCamelCase__    :   int					=				encodings['''input_ids''']
                 return inputs






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




           """simple docstring"""







           return self.tokenizer.batch_decode(*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )






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




           """simple docstring"""







           return self.tokenizer.decode(*__SCREAMING_SNAKE_CASE						,						**__SCREAMING_SNAKE_CASE  )






     @contextmanager
     def    __SCREAMING_SNAKE_CASE			(							self  )  ->   str:




           """simple docstring"""







           warnings.warn(
               '''`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your '''
               '''labels by using the argument `text` of the regular `__call__` method (either in the same call as '''
               '''your images inputs, or in a separate call.'''  )
           UpperCamelCase__    :   Any					=				True
           UpperCamelCase__    :   int					=				self.tokenizer
           yield
           UpperCamelCase__    :   str					=				self.image_processor
           UpperCamelCase__    :   Union[str, Any]					=				False






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




           """simple docstring"""







           if added_vocab is None:
                 UpperCamelCase__    :   List[Any]					=				self.tokenizer.get_added_vocab()

           UpperCamelCase__    :   Optional[int]					=				{}

           while tokens:
                 UpperCamelCase__    :   Optional[Any]					=				re.search(r'''<s_(.*?)>'''						,						__SCREAMING_SNAKE_CASE						,						re.IGNORECASE  )
                 if start_token is None:
                       break
                 UpperCamelCase__    :   Optional[Any]					=				start_token.group(1  )
                 UpperCamelCase__    :   List[str]					=				re.search(rF'''</s_{key}>'''						,						__SCREAMING_SNAKE_CASE						,						re.IGNORECASE  )
                 UpperCamelCase__    :   Dict					=				start_token.group()
                 if end_token is None:
                       UpperCamelCase__    :   Optional[int]					=				tokens.replace(__SCREAMING_SNAKE_CASE						,						''''''  )
                 else:
                       UpperCamelCase__    :   List[Any]					=				end_token.group()
                       UpperCamelCase__    :   List[Any]					=				re.escape(__SCREAMING_SNAKE_CASE  )
                       UpperCamelCase__    :   Optional[Any]					=				re.escape(__SCREAMING_SNAKE_CASE  )
                       UpperCamelCase__    :   Dict					=				re.search(F'''{start_token_escaped}(.*?){end_token_escaped}'''						,						__SCREAMING_SNAKE_CASE						,						re.IGNORECASE  )
                       if content is not None:
                             UpperCamelCase__    :   Any					=				content.group(1  ).strip()
                             if r"<s_" in content and r"</s_" in content:  # non-leaf node
                                   UpperCamelCase__    :   str					=				self.tokenajson(__SCREAMING_SNAKE_CASE						,						is_inner_value=__SCREAMING_SNAKE_CASE						,						added_vocab=__SCREAMING_SNAKE_CASE  )
                                   if value:
                                         if len(__SCREAMING_SNAKE_CASE  ) == 1:
                                               UpperCamelCase__    :   Union[str, Any]					=				value[0]
                                         UpperCamelCase__    :   str					=				value
                             else:  # leaf nodes
                                   UpperCamelCase__    :   Dict					=				[]
                                   for leaf in content.split(r'''<sep/>'''  ):
                                         UpperCamelCase__    :   int					=				leaf.strip()
                                         if leaf in added_vocab and leaf[0] == "<" and leaf[-2:] == "/>":
                                               UpperCamelCase__    :   Optional[Any]					=				leaf[1:-2]  # for categorical special tokens
                                         output[key].append(__SCREAMING_SNAKE_CASE  )
                                   if len(output[key]  ) == 1:
                                         UpperCamelCase__    :   Optional[int]					=				output[key][0]

                       UpperCamelCase__    :   Dict					=				tokens[tokens.find(__SCREAMING_SNAKE_CASE  ) + len(__SCREAMING_SNAKE_CASE  ) :].strip()
                       if tokens[:6] == r"<sep/>":  # non-leaf nodes
                             return [output] + self.tokenajson(tokens[6:]						,						is_inner_value=__SCREAMING_SNAKE_CASE						,						added_vocab=__SCREAMING_SNAKE_CASE  )

           if len(__SCREAMING_SNAKE_CASE  ):
                 return [output] if is_inner_value else output
           else:
                 return [] if is_inner_value else {"text_sequence": tokens}






     @property
     def    __SCREAMING_SNAKE_CASE			(							self  )  ->   Tuple:




           """simple docstring"""







           warnings.warn(
               '''`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'''						,						__SCREAMING_SNAKE_CASE						,						)
           return self.image_processor_class






     @property
     def    __SCREAMING_SNAKE_CASE			(							self  )  ->   Optional[int]:




           """simple docstring"""







           warnings.warn(
               '''`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'''						,						__SCREAMING_SNAKE_CASE						,						)
           return self.image_processor




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


