






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(__SCREAMING_SNAKE_CASE      )
class   lowerCamelCase				(__SCREAMING_SNAKE_CASE      ):







		'''simple docstring'''



		def __init__(							self    ,		*_UpperCamelCase    ,		**_UpperCamelCase       )     ->							Optional[int]:
							super().__init__(*_snake_case    ,		**_snake_case       )
							requires_backends(self    ,		'decord'       )
							self.check_model_type(_snake_case       )



		def  __UpperCAmelCase      (							self    ,		_UpperCamelCase=None    ,		_UpperCamelCase=None    ,		_UpperCamelCase=None       )     ->							List[str]:
							UpperCAmelCase_   :				Optional[Any]						=						{}
							if frame_sampling_rate is not None:
												UpperCAmelCase_   :				Optional[int]						=						frame_sampling_rate
							if num_frames is not None:
												UpperCAmelCase_   :				Union[str, Any]						=						num_frames

							UpperCAmelCase_   :				int						=						{}
							if top_k is not None:
												UpperCAmelCase_   :				Dict						=						top_k
							return preprocess_params, {}, postprocess_params



		def __call__(							self    ,		_UpperCamelCase    ,		**_UpperCamelCase       )     ->							Union[str, Any]:
							return super().__call__(_snake_case    ,		**_snake_case       )



		def  __UpperCAmelCase      (							self    ,		_UpperCamelCase    ,		_UpperCamelCase=None    ,		_UpperCamelCase=1       )     ->							Optional[Any]:
							if num_frames is None:
												UpperCAmelCase_   :				Any						=						self.model.config.num_frames

							if video.startswith('http://'       ) or video.startswith('https://'       ):
												UpperCAmelCase_   :				Optional[int]						=						BytesIO(requests.get(_snake_case       ).content       )

							UpperCAmelCase_   :				Tuple						=						VideoReader(_snake_case       )
							videoreader.seek(0       )

							UpperCAmelCase_   :				Dict						=						0
							UpperCAmelCase_   :				Optional[Any]						=						num_frames * frame_sampling_rate - 1
							UpperCAmelCase_   :				Dict						=						np.linspace(_snake_case    ,		_snake_case    ,		num=_snake_case    ,		dtype=np.intaa       )

							UpperCAmelCase_   :				int						=						videoreader.get_batch(_snake_case       ).asnumpy()
							UpperCAmelCase_   :				int						=						list(_snake_case       )

							UpperCAmelCase_   :				Any						=						self.image_processor(_snake_case    ,		return_tensors=self.framework       )
							return model_inputs



		def  __UpperCAmelCase      (							self    ,		_UpperCamelCase       )     ->							str:
							UpperCAmelCase_   :				int						=						self.model(**_snake_case       )
							return model_outputs



		def  __UpperCAmelCase      (							self    ,		_UpperCamelCase    ,		_UpperCamelCase=5       )     ->							Tuple:
							if top_k > self.model.config.num_labels:
												UpperCAmelCase_   :				List[str]						=						self.model.config.num_labels

							if self.framework == "pt":
												UpperCAmelCase_   :				str						=						model_outputs.logits.softmax(-1       )[0]
												UpperCAmelCase_   :				Optional[Any]						=						probs.topk(_snake_case       )
							else:
												raise ValueError(f"Unsupported framework: {self.framework}"       )

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













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   lowerCamelCase				(unittest.TestCase      ):







		'''simple docstring'''



		_snake_case   :   List[str]							=  MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING



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



		def  __UpperCAmelCase      (							self    ,		_UpperCamelCase    ,		_UpperCamelCase       )     ->							Dict:
							for example in examples:
												UpperCAmelCase_   :				str						=						video_classifier(_UpperCamelCase       )

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



		@require_torch
		def  __UpperCAmelCase      (							self       )     ->							List[Any]:
							UpperCAmelCase_   :				str						=						'hf-internal-testing/tiny-random-VideoMAEForVideoClassification'
							UpperCAmelCase_   :				Optional[Any]						=						VideoMAEFeatureExtractor(
							    size={'shortest_edge': 1_0}    ,		crop_size={'height': 1_0, 'width': 1_0}       )
							UpperCAmelCase_   :				str						=						pipeline(
							    'video-classification'    ,		model=_UpperCamelCase    ,		feature_extractor=_UpperCamelCase    ,		frame_sampling_rate=4       )

							UpperCAmelCase_   :				Any						=						hf_hub_download(repo_id='nateraw/video-demo'    ,		filename='archery.mp4'    ,		repo_type='dataset'       )
							UpperCAmelCase_   :				List[str]						=						video_classifier(_UpperCamelCase    ,		top_k=2       )
							self.assertEqual(
							    nested_simplify(_UpperCamelCase    ,		decimals=4       )    ,		[{'score': 0.51_99, 'label': 'LABEL_0'}, {'score': 0.48_01, 'label': 'LABEL_1'}]    ,		)

							UpperCAmelCase_   :				Tuple						=						video_classifier(
							    [
							        video_file_path,
							        video_file_path,
							    ]    ,		top_k=2    ,		)
							self.assertEqual(
							    nested_simplify(_UpperCamelCase    ,		decimals=4       )    ,		[
							        [{'score': 0.51_99, 'label': 'LABEL_0'}, {'score': 0.48_01, 'label': 'LABEL_1'}],
							        [{'score': 0.51_99, 'label': 'LABEL_0'}, {'score': 0.48_01, 'label': 'LABEL_1'}],
							    ]    ,		)



		@require_tf
		def  __UpperCAmelCase      (							self       )     ->							Dict:
							pass






