




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

lowercase      :   str      =       logging.get_logger(__name__)







@add_end_docstrings(a__	)
class 					_a (a__	):





     '''simple docstring'''



     def __init__(				self							,*__a							,**__a							)      ->				Tuple:
          super().__init__(*__a							,**__a							)
          requires_backends(self							,"""decord"""							)
          self.check_model_type(__a							)







     def 			snake_case_      (				self							,__a=None							,__a=None							,__a=None							)      ->				str:
          snake_case			:      str        =			{}
          if frame_sampling_rate is not None:
               snake_case			:      Union[str, Any]        =			frame_sampling_rate
          if num_frames is not None:
               snake_case			:      List[Any]        =			num_frames

          snake_case			:      Optional[Any]        =			{}
          if top_k is not None:
               snake_case			:      Optional[int]        =			top_k
          return preprocess_params, {}, postprocess_params







     def __call__(				self							,__a							,**__a							)      ->				Optional[int]:
          return super().__call__(__a							,**__a							)







     def 			snake_case_      (				self							,__a							,__a=None							,__a=1							)      ->				int:
          if num_frames is None:
               snake_case			:      Tuple        =			self.model.config.num_frames

          if video.startswith("""http://"""							) or video.startswith("""https://"""							):
               snake_case			:      Optional[Any]        =			BytesIO(requests.get(__a							).content							)

          snake_case			:      int        =			VideoReader(__a							)
          videoreader.seek(0							)

          snake_case			:      List[Any]        =			0
          snake_case			:      Union[str, Any]        =			num_frames * frame_sampling_rate - 1
          snake_case			:      Optional[Any]        =			np.linspace(__a							,__a							,num=__a							,dtype=np.intaa							)

          snake_case			:      List[Any]        =			videoreader.get_batch(__a							).asnumpy()
          snake_case			:      Dict        =			list(__a							)

          snake_case			:      Optional[Any]        =			self.image_processor(__a							,return_tensors=self.framework							)
          return model_inputs







     def 			snake_case_      (				self							,__a							)      ->				List[Any]:
          snake_case			:      List[Any]        =			self.model(**__a							)
          return model_outputs







     def 			snake_case_      (				self							,__a							,__a=5							)      ->				str:
          if top_k > self.model.config.num_labels:
               snake_case			:      List[str]        =			self.model.config.num_labels

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

          snake_case			:      Optional[Any]        =			scores.tolist()
          snake_case			:      str        =			ids.tolist()
          return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(__a							,__a							)]









'''simple docstring'''







import os


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_doctest_list.py
lowercase      :   List[str]      =       """."""


if __name__ == "__main__":
      lowercase      :   List[str]      =       os.path.join(REPO_PATH, """utils/documentation_tests.txt""")
      lowercase      :   Union[str, Any]      =       []
      lowercase      :   List[str]      =       []
      with open(doctest_file_path) as fp:
            for line in fp:
                  lowercase      :   List[str]      =       line.strip()
                  lowercase      :   Optional[Any]      =       os.path.join(REPO_PATH, line)
                  if not (os.path.isfile(path) or os.path.isdir(path)):
                        non_existent_paths.append(line)
                  all_paths.append(path)
      if len(non_existent_paths) > 0:
            lowercase      :   int      =       """\n""".join(non_existent_paths)
            raise ValueError(F"""`utils/documentation_tests.txt` contains non-existent paths:\n{non_existent_paths}""")
      if all_paths != sorted(all_paths):
            raise ValueError("""Files in `utils/documentation_tests.txt` are not in alphabetical order.""")



