




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_						=logging.get_logger(__name__)


@add_end_docstrings(__UpperCAmelCase  )
class 							__UpperCamelCase			( __UpperCAmelCase  ):





    '''simple docstring'''



    def __init__( self       , *UpperCAmelCase_       , **UpperCAmelCase_						):
           super().__init__(*UpperCAmelCase__       , **UpperCAmelCase__						)
           requires_backends(self       , '''decord'''						)
           self.check_model_type(UpperCAmelCase__						)



    def       __snake_case		( self       , UpperCAmelCase_=None       , UpperCAmelCase_=None       , UpperCAmelCase_=None						):
           lowerCAmelCase   =		{}
           if frame_sampling_rate is not None:
                  lowerCAmelCase   =		frame_sampling_rate
           if num_frames is not None:
                  lowerCAmelCase   =		num_frames

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



    def __call__( self       , UpperCAmelCase_       , **UpperCAmelCase_						):
           return super().__call__(UpperCAmelCase__       , **UpperCAmelCase__						)



    def       __snake_case		( self       , UpperCAmelCase_       , UpperCAmelCase_=None       , UpperCAmelCase_=1						):
           if num_frames is None:
                  lowerCAmelCase   =		self.model.config.num_frames

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

           lowerCAmelCase   =		VideoReader(UpperCAmelCase__						)
           videoreader.seek(0						)

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

           lowerCAmelCase   =		videoreader.get_batch(UpperCAmelCase__						).asnumpy()
           lowerCAmelCase   =		list(UpperCAmelCase__						)

           lowerCAmelCase   =		self.image_processor(UpperCAmelCase__       , return_tensors=self.framework						)
           return model_inputs



    def       __snake_case		( self       , UpperCAmelCase_						):
           lowerCAmelCase   =		self.model(**UpperCAmelCase__						)
           return model_outputs






    def       __snake_case		( self       , UpperCAmelCase_       , UpperCAmelCase_=5						):
           if top_k > self.model.config.num_labels:
                  lowerCAmelCase   =		self.model.config.num_labels

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

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









import unittest

from huggingface_hub import hf_hub_download

from transformers import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING, VideoMAEFeatureExtractor
from transformers.pipelines import VideoClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_decord,
    require_tf,
    require_torch,
    require_torch_or_tf,
    require_vision,
)

from .test_pipelines_common import ANY


@is_pipeline_test
@require_torch_or_tf
@require_vision
@require_decord
class 							__UpperCamelCase			( unittest.TestCase  ):





				'''simple docstring'''



				__a							: str										=MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING



				def       __snake_case		( self       , UpperCAmelCase_       , UpperCAmelCase_       , UpperCAmelCase_						):
											lowerCAmelCase   =		hf_hub_download(
											    repo_id='''nateraw/video-demo'''       , filename='''archery.mp4'''       , repo_type='''dataset'''						)
											lowerCAmelCase   =		VideoClassificationPipeline(model=UpperCAmelCase_       , image_processor=UpperCAmelCase_       , top_k=2						)
											lowerCAmelCase   =		[
											    example_video_filepath,
											    '''https://huggingface.co/datasets/nateraw/video-demo/resolve/main/archery.mp4''',
											]
											return video_classifier, examples



				def       __snake_case		( self       , UpperCAmelCase_       , UpperCAmelCase_						):
											for example in examples:
																		lowerCAmelCase   =		video_classifier(UpperCAmelCase_						)

																		self.assertEqual(
																		    UpperCAmelCase_       , [
																		        {'''score''': ANY(UpperCAmelCase_						), '''label''': ANY(UpperCAmelCase_						)},
																		        {'''score''': ANY(UpperCAmelCase_						), '''label''': ANY(UpperCAmelCase_						)},
																		    ]       , )



				@require_torch
				def       __snake_case		( self						):
											lowerCAmelCase   =		'''hf-internal-testing/tiny-random-VideoMAEForVideoClassification'''
											lowerCAmelCase   =		VideoMAEFeatureExtractor(
											    size={'''shortest_edge''': 10}       , crop_size={'''height''': 10, '''width''': 10}						)
											lowerCAmelCase   =		pipeline(
											    '''video-classification'''       , model=UpperCAmelCase_       , feature_extractor=UpperCAmelCase_       , frame_sampling_rate=4						)

											lowerCAmelCase   =		hf_hub_download(repo_id='''nateraw/video-demo'''       , filename='''archery.mp4'''       , repo_type='''dataset'''						)
											lowerCAmelCase   =		video_classifier(UpperCAmelCase_       , top_k=2						)
											self.assertEqual(
											    nested_simplify(UpperCAmelCase_       , decimals=4						)       , [{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}]       , )

											lowerCAmelCase   =		video_classifier(
											    [
											        video_file_path,
											        video_file_path,
											    ]       , top_k=2       , )
											self.assertEqual(
											    nested_simplify(UpperCAmelCase_       , decimals=4						)       , [
											        [{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}],
											        [{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}],
											    ]       , )






				@require_tf
				def       __snake_case		( self						):
											pass



