






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 							snake_case_       (    __UpperCamelCase      ):

						"""simple docstring"""



						def __init__(self:					List[str] ,					*__UpperCAmelCase:					Optional[Any] ,					**__UpperCAmelCase:					Optional[int]   )    ->     Dict:




											'''simple docstring'''



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



						def        UpperCAmelCase__					(self:					Dict ,					__UpperCAmelCase:					Tuple=None ,					__UpperCAmelCase:					Any=None ,					__UpperCAmelCase:					Optional[Any]=None   )    ->     Dict:




											'''simple docstring'''



											__a					:						List[str]      =					{}
											if frame_sampling_rate is not None:
																__a					:						List[Any]      =					frame_sampling_rate
											if num_frames is not None:
																__a					:						Dict      =					num_frames

											__a					:						Tuple      =					{}
											if top_k is not None:
																__a					:						List[str]      =					top_k
											return preprocess_params, {}, postprocess_params



						def __call__(self:					Optional[Any] ,					__UpperCAmelCase:					Union[str, List[str]] ,					**__UpperCAmelCase:					Any   )    ->     List[str]:




											'''simple docstring'''



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



						def        UpperCAmelCase__					(self:					List[Any] ,					__UpperCAmelCase:					Any ,					__UpperCAmelCase:					Tuple=None ,					__UpperCAmelCase:					List[Any]=1   )    ->     Optional[Any]:




											'''simple docstring'''



											if num_frames is None:
																__a					:						Tuple      =					self.model.config.num_frames

											if video.startswith("http://"   ) or video.startswith("https://"   ):
																__a					:						List[str]      =					BytesIO(requests.get(__UpperCAmelCase   ).content   )

											__a					:						Optional[Any]      =					VideoReader(__UpperCAmelCase   )
											videoreader.seek(0   )

											__a					:						Union[str, Any]      =					0
											__a					:						Any      =					num_frames * frame_sampling_rate - 1
											__a					:						Optional[Any]      =					np.linspace(__UpperCAmelCase ,					__UpperCAmelCase ,					num=__UpperCAmelCase ,					dtype=np.intaa   )

											__a					:						Tuple      =					videoreader.get_batch(__UpperCAmelCase   ).asnumpy()
											__a					:						int      =					list(__UpperCAmelCase   )

											__a					:						int      =					self.image_processor(__UpperCAmelCase ,					return_tensors=self.framework   )
											return model_inputs



						def        UpperCAmelCase__					(self:					Tuple ,					__UpperCAmelCase:					Optional[int]   )    ->     int:




											'''simple docstring'''



											__a					:						List[str]      =					self.model(**__UpperCAmelCase   )
											return model_outputs



						def        UpperCAmelCase__					(self:					List[Any] ,					__UpperCAmelCase:					List[Any] ,					__UpperCAmelCase:					Any=5   )    ->     Dict:




											'''simple docstring'''



											if top_k > self.model.config.num_labels:
																__a					:						str      =					self.model.config.num_labels

											if self.framework == "pt":
																__a					:						Optional[Any]      =					model_outputs.logits.softmax(-1   )[0]
																__a      ,			__a					:						int      =					probs.topk(__UpperCAmelCase   )
											else:
																raise ValueError(f'Unsupported framework: {self.framework}'   )

											__a					:						Union[str, Any]      =					scores.tolist()
											__a					:						Optional[int]      =					ids.tolist()
											return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(__UpperCAmelCase ,					__UpperCAmelCase   )]













from abc import ABC, abstractmethod
from argparse import ArgumentParser




class 							snake_case_       (    __UpperCamelCase      ):

						"""simple docstring"""



						@staticmethod
						@abstractmethod
						def        UpperCAmelCase__					(__UpperCAmelCase:					ArgumentParser   )    ->     Tuple:




											'''simple docstring'''



											raise NotImplementedError()



						@abstractmethod
						def        UpperCAmelCase__					(self:					List[str]   )    ->     List[str]:




											'''simple docstring'''



											raise NotImplementedError()





