





"""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

lowerCAmelCase__     			=			logging.get_logger(__name__)





@add_end_docstrings(UpperCamelCase_ )
class        _lowerCamelCase      (       UpperCamelCase_ ):


						def __init__(self   ,							*__a   ,							**__a		) ->   Optional[Any]:
								super().__init__(*__a   ,							**__a		)
								requires_backends(self   ,							"decord"		)
								self.check_model_type(__a		)


						def     snake_case_   (self   ,							__a=None   ,							__a=None   ,							__a=None		) ->   Any:
								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		) ->   List[Any]:
								return super().__call__(__a   ,							**__a		)


						def     snake_case_   (self   ,							__a   ,							__a=None   ,							__a=1		) ->   str:






								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		) ->   Dict:
								UpperCamelCase   =	self.model(**__a		)
								return model_outputs




						def     snake_case_   (self   ,							__a   ,							__a=5		) ->   List[str]:






								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   =	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		)]







"""simple docstring"""





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 ):
						UpperCAmelCase_ =			MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING


						def     snake_case_   (self   ,							__a   ,							__a   ,							__a		) ->   str:
								UpperCamelCase   =	hf_hub_download(
								    repo_id="nateraw/video-demo"   ,							filename="archery.mp4"   ,							repo_type="dataset"		)
								UpperCamelCase   =	VideoClassificationPipeline(model=__a   ,							image_processor=__a   ,							top_k=2		)
								UpperCamelCase   =	[
								    example_video_filepath,
								    "https://huggingface.co/datasets/nateraw/video-demo/resolve/main/archery.mp4",
								]
								return video_classifier, examples


						def     snake_case_   (self   ,							__a   ,							__a		) ->   str:
								for example in examples:
										UpperCamelCase   =	video_classifier(__a		)

										self.assertEqual(
										    __a   ,							[
										        {"score": ANY(__a		), "label": ANY(__a		)},
										        {"score": ANY(__a		), "label": ANY(__a		)},
										    ]   ,							)


						@require_torch
						def     snake_case_   (self		) ->   int:
								UpperCamelCase   =	"hf-internal-testing/tiny-random-VideoMAEForVideoClassification"
								UpperCamelCase   =	VideoMAEFeatureExtractor(
								    size={"shortest_edge": 10}   ,							crop_size={"height": 10, "width": 10}		)
								UpperCamelCase   =	pipeline(
								    "video-classification"   ,							model=__a   ,							feature_extractor=__a   ,							frame_sampling_rate=4		)

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

								UpperCamelCase   =	video_classifier(
								    [
								        video_file_path,
								        video_file_path,
								    ]   ,							top_k=2   ,							)
								self.assertEqual(
								    nested_simplify(__a   ,							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		) ->   Optional[Any]:
								pass






