
'''simple docstring'''



import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import is_speech_available, is_vision_available
from transformers.testing_utils import require_torch


if is_vision_available():
     from transformers import TvltImageProcessor

if is_speech_available():
     from transformers import TvltFeatureExtractor

from transformers import TvltProcessor

@require_torch
class  _a					(unittest.TestCase):





       """simple docstring"""




       def 							UpperCamelCase					(				self   )     ->      Tuple:
          _SCREAMING_SNAKE_CASE  			=					"""ZinengTang/tvlt-base"""
          _SCREAMING_SNAKE_CASE  			=					tempfile.mkdtemp()






       def 							UpperCamelCase					(				self							,				**A__   )     ->      List[Any]:
          return TvltImageProcessor.from_pretrained(self.checkpoint							,				**a_   )






       def 							UpperCamelCase					(				self							,				**A__   )     ->      List[Any]:
          return TvltFeatureExtractor.from_pretrained(self.checkpoint							,				**a_   )






       def 							UpperCamelCase					(				self   )     ->      List[str]:
          shutil.rmtree(self.tmpdirname   )






       def 							UpperCamelCase					(				self   )     ->      Any:
          _SCREAMING_SNAKE_CASE  			=					self.get_image_processor()
          _SCREAMING_SNAKE_CASE  			=					self.get_feature_extractor()

          _SCREAMING_SNAKE_CASE  			=					TvltProcessor(image_processor=a_							,				feature_extractor=a_   )
          processor.save_pretrained(self.tmpdirname   )
          _SCREAMING_SNAKE_CASE  			=					TvltProcessor.from_pretrained(self.tmpdirname   )

          self.assertIsInstance(processor.feature_extractor							,				a_   )
          self.assertIsInstance(processor.image_processor							,				a_   )






       def 							UpperCamelCase					(				self   )     ->      Any:
          _SCREAMING_SNAKE_CASE  			=					self.get_image_processor()
          _SCREAMING_SNAKE_CASE  			=					self.get_feature_extractor()

          _SCREAMING_SNAKE_CASE  			=					TvltProcessor(image_processor=a_							,				feature_extractor=a_   )

          _SCREAMING_SNAKE_CASE  			=					np.ones([1_20_00]   )

          _SCREAMING_SNAKE_CASE  			=					feature_extractor(a_							,				return_tensors="""np"""   )
          _SCREAMING_SNAKE_CASE  			=					processor(audio=a_							,				return_tensors="""np"""   )

          for key in audio_dict.keys():
             self.assertAlmostEqual(audio_dict[key].sum()							,				input_processor[key].sum()							,				delta=1E-2   )






       def 							UpperCamelCase					(				self   )     ->      Union[str, Any]:
          _SCREAMING_SNAKE_CASE  			=					self.get_image_processor()
          _SCREAMING_SNAKE_CASE  			=					self.get_feature_extractor()

          _SCREAMING_SNAKE_CASE  			=					TvltProcessor(image_processor=a_							,				feature_extractor=a_   )

          _SCREAMING_SNAKE_CASE  			=					np.ones([3, 2_24, 2_24]   )

          _SCREAMING_SNAKE_CASE  			=					image_processor(a_							,				return_tensors="""np"""   )
          _SCREAMING_SNAKE_CASE  			=					processor(images=a_							,				return_tensors="""np"""   )

          for key in image_dict.keys():
             self.assertAlmostEqual(image_dict[key].sum()							,				input_processor[key].sum()							,				delta=1E-2   )






       def 							UpperCamelCase					(				self   )     ->      Dict:
          _SCREAMING_SNAKE_CASE  			=					self.get_image_processor()
          _SCREAMING_SNAKE_CASE  			=					self.get_feature_extractor()

          _SCREAMING_SNAKE_CASE  			=					TvltProcessor(image_processor=a_							,				feature_extractor=a_   )

          _SCREAMING_SNAKE_CASE  			=					np.ones([1_20_00]   )
          _SCREAMING_SNAKE_CASE  			=					np.ones([3, 2_24, 2_24]   )

          _SCREAMING_SNAKE_CASE  			=					processor(audio=a_							,				images=a_   )

          self.assertListEqual(list(inputs.keys()   )							,				["""audio_values""", """audio_mask""", """pixel_values""", """pixel_mask"""]   )

          # test if it raises when no input is passed
          with pytest.raises(a_   ):
             processor()






       def 							UpperCamelCase					(				self   )     ->      int:
          _SCREAMING_SNAKE_CASE  			=					self.get_image_processor()
          _SCREAMING_SNAKE_CASE  			=					self.get_feature_extractor()

          _SCREAMING_SNAKE_CASE  			=					TvltProcessor(image_processor=a_							,				feature_extractor=a_   )

          self.assertListEqual(
              processor.model_input_names							,				image_processor.model_input_names + feature_extractor.model_input_names							,				msg="""`processor` and `image_processor`+`feature_extractor` model input names do not match"""							,				)









'''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					:			List[Any]							=    logging.get_logger(__name__)



@add_end_docstrings(a			)
class 							lowercase_						(		a			):

  '''simple docstring'''





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


    """simple docstring"""
    super().__init__(*a_ ,			**a_  )
    requires_backends(self ,			'decord'  )
    self.check_model_type(a_  )
  def     snake_case_      (  self ,			a_=None ,			a_=None ,			a_=None  )				->					int:


    """simple docstring"""
    UpperCAmelCase    				=						{}
    if frame_sampling_rate is not None:
      UpperCAmelCase    				=						frame_sampling_rate
    if num_frames is not None:
      UpperCAmelCase    				=						num_frames

    UpperCAmelCase    				=						{}
    if top_k is not None:
      UpperCAmelCase    				=						top_k
    return preprocess_params, {}, postprocess_params
  def __call__(  self ,			a_ ,			**a_  )				->					Union[str, Any]:


    """simple docstring"""
    return super().__call__(a_ ,			**a_  )
  def     snake_case_      (  self ,			a_ ,			a_=None ,			a_=1  )				->					Tuple:


    """simple docstring"""
    if num_frames is None:
      UpperCAmelCase    				=						self.model.config.num_frames

    if video.startswith('http://'  ) or video.startswith('https://'  ):
      UpperCAmelCase    				=						BytesIO(requests.get(a_  ).content  )

    UpperCAmelCase    				=						VideoReader(a_  )
    videoreader.seek(0  )

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

    UpperCAmelCase    				=						videoreader.get_batch(a_  ).asnumpy()
    UpperCAmelCase    				=						list(a_  )

    UpperCAmelCase    				=						self.image_processor(a_ ,			return_tensors=self.framework  )
    return model_inputs
  def     snake_case_      (  self ,			a_  )				->					Union[str, Any]:


    """simple docstring"""
    UpperCAmelCase    				=						self.model(**a_  )
    return model_outputs





  def     snake_case_      (  self ,			a_ ,			a_=5  )				->					Union[str, Any]:


    """simple docstring"""
    if top_k > self.model.config.num_labels:
      UpperCAmelCase    				=						self.model.config.num_labels

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

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


