





"""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(_lowercase )
class        _lowerCamelCase      (       _lowercase ):


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


						def     snake_case_   (self   ,							__a=None   ,							__a=None   ,							__a=None		) ->   Tuple:
								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		) ->   str:
								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		) ->   Any:
								UpperCamelCase   =	self.model(**__a		)
								return model_outputs




						def     snake_case_   (self   ,							__a   ,							__a=5		) ->   Any:
								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		)]













"""simple docstring"""





import pytest

from datasets.parallel import ParallelBackendConfig, parallel_backend
from datasets.utils.py_utils import map_nested

from .utils import require_dill_gt_0_3_2, require_joblibspark, require_not_windows


def       a__					(					_SCREAMING_SNAKE_CASE    ):  # picklable for multiprocessing




		"""simple docstring"""


		return i + 1


@require_dill_gt_0_3_2
@require_joblibspark
@require_not_windows
def       a__					(					):




		"""simple docstring"""


		with parallel_backend("spark"    ):
				assert ParallelBackendConfig.backend_name == "spark"

		UpperCamelCase   =	[1, 2, 3]
		with pytest.raises(_SCREAMING_SNAKE_CASE    ):
				with parallel_backend("unsupported backend"    ):
						map_nested(_SCREAMING_SNAKE_CASE  ,	_SCREAMING_SNAKE_CASE  ,	num_proc=2    )

		with pytest.raises(_SCREAMING_SNAKE_CASE    ):
				with parallel_backend("unsupported backend"    ):
						map_nested(_SCREAMING_SNAKE_CASE  ,	_SCREAMING_SNAKE_CASE  ,	num_proc=-1    )






@require_dill_gt_0_3_2
@require_joblibspark
@require_not_windows
@pytest.mark.parametrize("num_proc"  ,	[2, -1]    )
def       a__					(					_SCREAMING_SNAKE_CASE    ):




		"""simple docstring"""


		UpperCamelCase   =	[1, 2]
		UpperCamelCase   =	{"a": 1, "b": 2}
		UpperCamelCase   =	{"a": [1, 2], "b": [3, 4]}
		UpperCamelCase   =	{"a": {"1": 1}, "b": 2}
		UpperCamelCase   =	{"a": 1, "b": 2, "c": 3, "d": 4}
		UpperCamelCase   =	[2, 3]
		UpperCamelCase   =	{"a": 2, "b": 3}
		UpperCamelCase   =	{"a": [2, 3], "b": [4, 5]}
		UpperCamelCase   =	{"a": {"1": 2}, "b": 3}
		UpperCamelCase   =	{"a": 2, "b": 3, "c": 4, "d": 5}

		with parallel_backend("spark"    ):
				assert map_nested(_SCREAMING_SNAKE_CASE  ,	_SCREAMING_SNAKE_CASE  ,	num_proc=_SCREAMING_SNAKE_CASE    ) == expected_map_nested_sa
				assert map_nested(_SCREAMING_SNAKE_CASE  ,	_SCREAMING_SNAKE_CASE  ,	num_proc=_SCREAMING_SNAKE_CASE    ) == expected_map_nested_sa
				assert map_nested(_SCREAMING_SNAKE_CASE  ,	_SCREAMING_SNAKE_CASE  ,	num_proc=_SCREAMING_SNAKE_CASE    ) == expected_map_nested_sa
				assert map_nested(_SCREAMING_SNAKE_CASE  ,	_SCREAMING_SNAKE_CASE  ,	num_proc=_SCREAMING_SNAKE_CASE    ) == expected_map_nested_sa
				assert map_nested(_SCREAMING_SNAKE_CASE  ,	_SCREAMING_SNAKE_CASE  ,	num_proc=_SCREAMING_SNAKE_CASE    ) == expected_map_nested_sa






