

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

UpperCAmelCase__ :     Union[str, Any]				    =    logging.get_logger(__name__)







@add_end_docstrings(UpperCAmelCase       )
class        a__       (   UpperCAmelCase       ):







  """simple docstring"""




  def __init__(			self    : Any				,				*UpperCAmelCase__    : Union[str, Any]				,				**UpperCAmelCase__    : Dict		)					->Optional[int]:
        """simple docstring"""







        super().__init__(*UpperCAmelCase__				,				**UpperCAmelCase__		)
        requires_backends(self				,				"""decord"""		)
        self.check_model_type(UpperCAmelCase__		)


  def  _lowercase      (			self    : str				,				UpperCAmelCase__    : Union[str, Any]=None				,				UpperCAmelCase__    : List[Any]=None				,				UpperCAmelCase__    : Tuple=None		)					->Dict:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Any       =   {}
        if frame_sampling_rate is not None:
              SCREAMING_SNAKE_CASE :      Union[str, Any]       =   frame_sampling_rate
        if num_frames is not None:
              SCREAMING_SNAKE_CASE :      Optional[Any]       =   num_frames

        SCREAMING_SNAKE_CASE :      str       =   {}
        if top_k is not None:
              SCREAMING_SNAKE_CASE :      List[str]       =   top_k
        return preprocess_params, {}, postprocess_params


  def __call__(			self    : Union[str, Any]				,				UpperCAmelCase__    : Union[str, List[str]]				,				**UpperCAmelCase__    : int		)					->Optional[Any]:
        """simple docstring"""







        return super().__call__(UpperCAmelCase__				,				**UpperCAmelCase__		)


  def  _lowercase      (			self    : Union[str, Any]				,				UpperCAmelCase__    : Any				,				UpperCAmelCase__    : Any=None				,				UpperCAmelCase__    : List[Any]=1		)					->Optional[int]:
        """simple docstring"""







        if num_frames is None:
              SCREAMING_SNAKE_CASE :      int       =   self.model.config.num_frames

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

        SCREAMING_SNAKE_CASE :      List[Any]       =   VideoReader(UpperCAmelCase__		)
        videoreader.seek(0		)

        SCREAMING_SNAKE_CASE :      Tuple       =   0
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   num_frames * frame_sampling_rate - 1
        SCREAMING_SNAKE_CASE :      Tuple       =   np.linspace(UpperCAmelCase__				,				UpperCAmelCase__				,				num=UpperCAmelCase__				,				dtype=np.intaa		)

        SCREAMING_SNAKE_CASE :      str       =   videoreader.get_batch(UpperCAmelCase__		).asnumpy()
        SCREAMING_SNAKE_CASE :      str       =   list(UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      Tuple       =   self.image_processor(UpperCAmelCase__				,				return_tensors=self.framework		)
        return model_inputs


  def  _lowercase      (			self    : List[str]				,				UpperCAmelCase__    : Optional[int]		)					->Tuple:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Dict       =   self.model(**UpperCAmelCase__		)
        return model_outputs




  def  _lowercase      (			self    : int				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : int=5		)					->int:
        """simple docstring"""







        if top_k > self.model.config.num_labels:
              SCREAMING_SNAKE_CASE :      Any       =   self.model.config.num_labels

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

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







from __future__ import annotations

import numpy as np






def 					__lowercase      (     _A			)     ->						tuple[np.ndarray, np.ndarray]:
      SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      Dict       =   np.shape(_A			)
      if rows != columns:
            SCREAMING_SNAKE_CASE :      int       =   (
                """'table' has to be of square shaped array but got a """
                F"{rows}x{columns} array:\n{table}"
            )
            raise ValueError(_A			)

      SCREAMING_SNAKE_CASE :      Union[str, Any]       =   np.zeros((rows, columns)			)
      SCREAMING_SNAKE_CASE :      Optional[Any]       =   np.zeros((rows, columns)			)
      for i in range(_A			):
            for j in range(_A			):
                  SCREAMING_SNAKE_CASE :      Tuple       =   sum(lower[i][k] * upper[k][j] for k in range(_A			)			)
                  if upper[j][j] == 0:
                        raise ArithmeticError("""No LU decomposition exists"""			)
                  SCREAMING_SNAKE_CASE :      Union[str, Any]       =   (table[i][j] - total) / upper[j][j]
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   1
            for j in range(_A						,    _A			):
                  SCREAMING_SNAKE_CASE :      List[str]       =   sum(lower[i][k] * upper[k][j] for k in range(_A			)			)
                  SCREAMING_SNAKE_CASE :      Optional[Any]       =   table[i][j] - total
      return lower, upper


if __name__ == "__main__":
  import doctest

  doctest.testmod()




