






'''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(UpperCamelCase_							)
class a_    (    UpperCamelCase_							):







   def __init__(self					,   *__a					,   **__a)			->      Optional[Any]:
      """simple docstring"""






      super().__init__(*__a					,   **__a)
      requires_backends(self					,   'decord')
      self.check_model_type(__a)







   def     SCREAMING_SNAKE_CASE__					(self					,   __a=None					,   __a=None					,   __a=None)			->      Any:
      """simple docstring"""






      __snake_case				:      Tuple     							= {}
      if frame_sampling_rate is not None:
         __snake_case				:      Optional[int]     							= frame_sampling_rate
      if num_frames is not None:
         __snake_case				:      Dict     							= num_frames

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







   def __call__(self					,   __a					,   **__a)			->      List[Any]:
      """simple docstring"""






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







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a=None					,   __a=1)			->      str:
      """simple docstring"""






      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				:      Dict     							= VideoReader(__a)
      videoreader.seek(0)

      __snake_case				:      Dict     							= 0
      __snake_case				:      Dict     							= num_frames * frame_sampling_rate - 1
      __snake_case				:      List[str]     							= np.linspace(__a					,   __a					,   num=__a					,   dtype=np.intaa)

      __snake_case				:      str     							= videoreader.get_batch(__a).asnumpy()
      __snake_case				:      List[str]     							= list(__a)

      __snake_case				:      Tuple     							= self.image_processor(__a					,   return_tensors=self.framework)
      return model_inputs







   def     SCREAMING_SNAKE_CASE__					(self					,   __a)			->      Dict:
      """simple docstring"""






      __snake_case				:      Union[str, Any]     							= self.model(**__a)
      return model_outputs







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a=5)			->      List[str]:
      """simple docstring"""






      if top_k > self.model.config.num_labels:
         __snake_case				:      Optional[int]     							= self.model.config.num_labels

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

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







'''simple docstring'''






import unittest

from transformers import load_tool
from transformers.utils import is_torch_available


if is_torch_available():
      import torch

from transformers.testing_utils import require_torch

from .test_tools_common import ToolTesterMixin




@require_torch
class a_    (    unittest.TestCase		,      UpperCamelCase_							):







   def     SCREAMING_SNAKE_CASE__					(self)			->      List[Any]:
      """simple docstring"""






      __snake_case				:      List[str]     							= load_tool('text-to-speech')
      self.tool.setup()







   def     SCREAMING_SNAKE_CASE__					(self)			->      Optional[Any]:
      """simple docstring"""






      torch.manual_seed(0)
      __snake_case				:      Dict     							= self.tool('hey')
      __snake_case				:      List[Any]     							= result.to_raw()
      self.assertTrue(
          torch.allclose(
              resulting_tensor[:3]					,   torch.tensor([-0.0_005_966_668_832_115_829, -0.0_003_657_640_190_795_064, -0.00_013_439_502_799_883_485])					,   ))







   def     SCREAMING_SNAKE_CASE__					(self)			->      List[Any]:
      """simple docstring"""






      torch.manual_seed(0)
      __snake_case				:      Any     							= self.tool('hey')
      __snake_case				:      Any     							= result.to_raw()
      self.assertTrue(
          torch.allclose(
              resulting_tensor[:3]					,   torch.tensor([-0.0_005_966_668_832_115_829, -0.0_003_657_640_190_795_064, -0.00_013_439_502_799_883_485])					,   ))