

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

__lowerCamelCase						    =      logging.get_logger(__name__)


@add_end_docstrings(_snake_case     )
class    A__ ( _snake_case     ):



    def __init__(   self						,			*UpperCamelCase__						,			**UpperCamelCase__    )  -> str:





           '''simple docstring'''
           super().__init__(*UpperCamelCase__						,			**UpperCamelCase__    )
           requires_backends(self						,			"""decord"""    )
           self.check_model_type(UpperCamelCase__    )



    def    snake_case_     (   self						,			UpperCamelCase__=None						,			UpperCamelCase__=None						,			UpperCamelCase__=None    )  -> Any:





           '''simple docstring'''
           A_		   = {}
           if frame_sampling_rate is not None:
                  A_		   = frame_sampling_rate
           if num_frames is not None:
                  A_		   = num_frames

           A_		   = {}
           if top_k is not None:
                  A_		   = top_k
           return preprocess_params, {}, postprocess_params



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





           '''simple docstring'''
           return super().__call__(UpperCamelCase__						,			**UpperCamelCase__    )



    def    snake_case_     (   self						,			UpperCamelCase__						,			UpperCamelCase__=None						,			UpperCamelCase__=1    )  -> Optional[int]:





           '''simple docstring'''
           if num_frames is None:
                  A_		   = self.model.config.num_frames

           if video.startswith("""http://"""    ) or video.startswith("""https://"""    ):
                  A_		   = BytesIO(requests.get(UpperCamelCase__    ).content    )

           A_		   = VideoReader(UpperCamelCase__    )
           videoreader.seek(0    )

           A_		   = 0
           A_		   = num_frames * frame_sampling_rate - 1
           A_		   = np.linspace(UpperCamelCase__						,			UpperCamelCase__						,			num=UpperCamelCase__						,			dtype=np.intaa    )

           A_		   = videoreader.get_batch(UpperCamelCase__    ).asnumpy()
           A_		   = list(UpperCamelCase__    )

           A_		   = self.image_processor(UpperCamelCase__						,			return_tensors=self.framework    )
           return model_inputs



    def    snake_case_     (   self						,			UpperCamelCase__    )  -> Dict:





           '''simple docstring'''
           A_		   = self.model(**UpperCamelCase__    )
           return model_outputs







    def    snake_case_     (   self						,			UpperCamelCase__						,			UpperCamelCase__=5    )  -> List[Any]:





           '''simple docstring'''
           if top_k > self.model.config.num_labels:
                  A_		   = self.model.config.num_labels

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

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



'''simple docstring'''







from functools import lru_cache
@lru_cache
def  UpperCAmelCase__      ( UpperCAmelCase__			)     ->     int:
       if num < 0:
              raise ValueError("""Number should not be negative."""			)

       return 1 if num in (0, 1) else num * factorial(num - 1			)


if __name__ == "__main__":
      import doctest

      doctest.testmod()
