

'''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

A_       						=							logging.get_logger(__name__)





@add_end_docstrings(_a						)
class 							_snake_case (       _a						):
   def __init__( self			:      Optional[Any]     ,*SCREAMING_SNAKE_CASE__			:      str     ,**SCREAMING_SNAKE_CASE__			:      List[str]							):
     super().__init__(*SCREAMING_SNAKE_CASE__     ,**SCREAMING_SNAKE_CASE__							)
     requires_backends(self     ,"decord"							)
     self.check_model_type(SCREAMING_SNAKE_CASE__							)
   def     __UpperCamelCase						( self			:      Union[str, Any]     ,SCREAMING_SNAKE_CASE__			:      Optional[Any]=None     ,SCREAMING_SNAKE_CASE__			:      Dict=None     ,SCREAMING_SNAKE_CASE__			:      Union[str, Any]=None							):
     SCREAMING_SNAKE_CASE:int						      =					{}
     if frame_sampling_rate is not None:
       SCREAMING_SNAKE_CASE:List[str]						      =					frame_sampling_rate
     if num_frames is not None:
       SCREAMING_SNAKE_CASE:int						      =					num_frames

     SCREAMING_SNAKE_CASE:List[Any]						      =					{}
     if top_k is not None:
       SCREAMING_SNAKE_CASE:Union[str, Any]						      =					top_k
     return preprocess_params, {}, postprocess_params
   def __call__( self			:      List[Any]     ,SCREAMING_SNAKE_CASE__			:      Union[str, List[str]]     ,**SCREAMING_SNAKE_CASE__			:      Dict							):
     return super().__call__(SCREAMING_SNAKE_CASE__     ,**SCREAMING_SNAKE_CASE__							)
   def     __UpperCamelCase						( self			:      List[Any]     ,SCREAMING_SNAKE_CASE__			:      Union[str, Any]     ,SCREAMING_SNAKE_CASE__			:      Tuple=None     ,SCREAMING_SNAKE_CASE__			:      Union[str, Any]=1							):
     if num_frames is None:
       SCREAMING_SNAKE_CASE:Any						      =					self.model.config.num_frames

     if video.startswith("http://"							) or video.startswith("https://"							):
       SCREAMING_SNAKE_CASE:str						      =					BytesIO(requests.get(SCREAMING_SNAKE_CASE__							).content							)

     SCREAMING_SNAKE_CASE:Dict						      =					VideoReader(SCREAMING_SNAKE_CASE__							)
     videoreader.seek(0							)

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

     SCREAMING_SNAKE_CASE:Optional[int]						      =					videoreader.get_batch(SCREAMING_SNAKE_CASE__							).asnumpy()
     SCREAMING_SNAKE_CASE:Any						      =					list(SCREAMING_SNAKE_CASE__							)

     SCREAMING_SNAKE_CASE:List[Any]						      =					self.image_processor(SCREAMING_SNAKE_CASE__     ,return_tensors=self.framework							)
     return model_inputs
   def     __UpperCamelCase						( self			:      Tuple     ,SCREAMING_SNAKE_CASE__			:      Optional[Any]							):
     SCREAMING_SNAKE_CASE:Optional[int]						      =					self.model(**SCREAMING_SNAKE_CASE__							)
     return model_outputs







   def     __UpperCamelCase						( self			:      Any     ,SCREAMING_SNAKE_CASE__			:      Optional[int]     ,SCREAMING_SNAKE_CASE__			:      List[str]=5							):
     if top_k > self.model.config.num_labels:
       SCREAMING_SNAKE_CASE:Dict						      =					self.model.config.num_labels

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

     SCREAMING_SNAKE_CASE:Dict						      =					scores.tolist()
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					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 warnings
from functools import wraps
from typing import Callable





def    A_  ( snake_case ):

  @wraps(snake_case )
  def _inner_fn(*snake_case      ,       **snake_case ):
    warnings.warn(
        (F'''\'{fn.__name__}\' is experimental and might be subject to breaking changes in the future.''')      ,       snake_case      ,       )
    return fn(*snake_case      ,       **snake_case )

  return _inner_fn






