



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

__snake_case       :	Tuple            =logging.get_logger(__name__)



@add_end_docstrings(lowerCamelCase__)
class  lowerCamelCase__    (							lowerCamelCase__):




		'''simple docstring'''



		def __init__(self	,*__lowerCamelCase	,**__lowerCamelCase				)  ->    List[str]:

						"""simple docstring"""






						super().__init__(*UpperCAmelCase__	,**UpperCAmelCase__				)
						requires_backends(self	,'''decord'''				)
						self.check_model_type(UpperCAmelCase__				)



		def 	lowerCAmelCase__   (self	,__lowerCamelCase=None	,__lowerCamelCase=None	,__lowerCamelCase=None				)  ->    str:

						"""simple docstring"""






						lowerCAmelCase__  :       Union[str, Any]				=  {}
						if frame_sampling_rate is not None:
										lowerCAmelCase__  :       List[str]				=  frame_sampling_rate
						if num_frames is not None:
										lowerCAmelCase__  :       Any				=  num_frames

						lowerCAmelCase__  :       Optional[Any]				=  {}
						if top_k is not None:
										lowerCAmelCase__  :       Any				=  top_k
						return preprocess_params, {}, postprocess_params



		def __call__(self	,__lowerCamelCase	,**__lowerCamelCase				)  ->    int:

						"""simple docstring"""






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



		def 	lowerCAmelCase__   (self	,__lowerCamelCase	,__lowerCamelCase=None	,__lowerCamelCase=1				)  ->    str:

						"""simple docstring"""






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

						if video.startswith('''http://'''				) or video.startswith('''https://'''				):
										lowerCAmelCase__  :       List[Any]				=  BytesIO(requests.get(UpperCAmelCase__				).content				)

						lowerCAmelCase__  :       Optional[Any]				=  VideoReader(UpperCAmelCase__				)
						videoreader.seek(0				)

						lowerCAmelCase__  :       Dict				=  0
						lowerCAmelCase__  :       Dict				=  num_frames * frame_sampling_rate - 1
						lowerCAmelCase__  :       List[str]				=  np.linspace(UpperCAmelCase__	,UpperCAmelCase__	,num=UpperCAmelCase__	,dtype=np.intaa				)

						lowerCAmelCase__  :       List[Any]				=  videoreader.get_batch(UpperCAmelCase__				).asnumpy()
						lowerCAmelCase__  :       int				=  list(UpperCAmelCase__				)

						lowerCAmelCase__  :       Tuple				=  self.image_processor(UpperCAmelCase__	,return_tensors=self.framework				)
						return model_inputs



		def 	lowerCAmelCase__   (self	,__lowerCamelCase				)  ->    Tuple:

						"""simple docstring"""






						lowerCAmelCase__  :       Optional[Any]				=  self.model(**UpperCAmelCase__				)
						return model_outputs



		def 	lowerCAmelCase__   (self	,__lowerCamelCase	,__lowerCamelCase=5				)  ->    Tuple:

						"""simple docstring"""






						if top_k > self.model.config.num_labels:
										lowerCAmelCase__  :       Any				=  self.model.config.num_labels

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

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




from collections.abc import Callable

import numpy as np






def 			lowerCAmelCase__	(					lowerCamelCase_	:	Callable ,lowerCamelCase_	:	float ,lowerCamelCase_	:	float ,lowerCamelCase_	:	float ,lowerCamelCase_	:	float):





				'''simple docstring'''

				lowerCAmelCase__  :       Dict				=  int(np.ceil((x_end - xa) / step_size))
				lowerCAmelCase__  :       Any				=  np.zeros((n + 1,))
				lowerCAmelCase__  :       Union[str, Any]				=  ya
				lowerCAmelCase__  :       str				=  xa

				for k in range(lowerCamelCase_):
								lowerCAmelCase__  :       Tuple				=  y[k] + step_size * ode_func(lowerCamelCase_ ,y[k])
								lowerCAmelCase__  :       str				=  y[k] + (
								    (step_size / 2) * (ode_func(lowerCamelCase_ ,y[k]) + ode_func(x + step_size ,lowerCamelCase_))
								)
								x += step_size

				return y


if __name__ == "__main__":
							import doctest

							doctest.testmod()

