





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

__magic_name__       :					List[Any]						=			logging.get_logger(__name__)






@add_end_docstrings(__SCREAMING_SNAKE_CASE	)
class 							lowercase__			(   __SCREAMING_SNAKE_CASE	):

			"""simple docstring"""







			def __init__(      self						,   *_A						,   **_A    ):






									'''simple docstring'''







									super().__init__(*_A						,   **_A    )
									requires_backends(self						,   """decord"""    )
									self.check_model_type(_A    )







			def   _a      (      self						,   _A=None						,   _A=None						,   _A=None    ):






									'''simple docstring'''







									UpperCamelCase				:		Any				   =			{}
									if frame_sampling_rate is not None:
															UpperCamelCase				:		int				   =			frame_sampling_rate
									if num_frames is not None:
															UpperCamelCase				:		List[Any]				   =			num_frames

									UpperCamelCase				:		int				   =			{}
									if top_k is not None:
															UpperCamelCase				:		List[Any]				   =			top_k
									return preprocess_params, {}, postprocess_params







			def __call__(      self						,   _A						,   **_A    ):






									'''simple docstring'''







									return super().__call__(_A						,   **_A    )







			def   _a      (      self						,   _A						,   _A=None						,   _A=1    ):






									'''simple docstring'''







									if num_frames is None:
															UpperCamelCase				:		Optional[int]				   =			self.model.config.num_frames

									if video.startswith("""http://"""    ) or video.startswith("""https://"""    ):
															UpperCamelCase				:		Optional[int]				   =			BytesIO(requests.get(_A    ).content    )

									UpperCamelCase				:		Any				   =			VideoReader(_A    )
									videoreader.seek(0    )

									UpperCamelCase				:		Optional[int]				   =			0
									UpperCamelCase				:		int				   =			num_frames * frame_sampling_rate - 1
									UpperCamelCase				:		List[Any]				   =			np.linspace(_A						,   _A						,   num=_A						,   dtype=np.intaa    )

									UpperCamelCase				:		List[Any]				   =			videoreader.get_batch(_A    ).asnumpy()
									UpperCamelCase				:		Any				   =			list(_A    )

									UpperCamelCase				:		Dict				   =			self.image_processor(_A						,   return_tensors=self.framework    )
									return model_inputs







			def   _a      (      self						,   _A    ):






									'''simple docstring'''







									UpperCamelCase				:		str				   =			self.model(**_A    )
									return model_outputs







			def   _a      (      self						,   _A						,   _A=5    ):






									'''simple docstring'''







									if top_k > self.model.config.num_labels:
															UpperCamelCase				:		Optional[Any]				   =			self.model.config.num_labels

									if self.framework == "pt":
															UpperCamelCase				:		List[str]				   =			model_outputs.logits.softmax(-1    )[0]
															UpperCamelCase   ,						UpperCamelCase				:		Union[str, Any]				   =			probs.topk(_A    )
									else:
															raise ValueError(f"""Unsupported framework: {self.framework}"""    )

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










"""simple docstring"""


from ...utils import is_torch_available, is_transformers_available


if is_transformers_available() and is_torch_available():
		from .pipeline_vq_diffusion import LearnedClassifierFreeSamplingEmbeddings, VQDiffusionPipeline



