




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

A_						=				logging.get_logger(__name__)






@add_end_docstrings(__a )
class 					lowercase(     __a ):







 '''simple docstring'''





 def __init__(				self:		Optional[Any],      *a_:		Dict,      **a_:		int		):




   '''simple docstring'''



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


 def  UpperCamelCase_   (				self:		Union[str, Any],      a_:		str=None,      a_:		Any=None,      a_:		Optional[int]=None		):




   '''simple docstring'''



   _snake_case							:		List[str]								=  {}
   if frame_sampling_rate is not None:
     _snake_case							:		str								=  frame_sampling_rate
   if num_frames is not None:
     _snake_case							:		List[str]								=  num_frames

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


 def __call__(				self:		Union[str, Any],      a_:		Union[str, List[str]],      **a_:		Tuple		):




   '''simple docstring'''



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


 def  UpperCamelCase_   (				self:		Union[str, Any],      a_:		List[str],      a_:		Optional[int]=None,      a_:		Optional[int]=1		):




   '''simple docstring'''



   if num_frames is None:
     _snake_case							:		Optional[Any]								=  self.model.config.num_frames

   if video.startswith("""http://"""		) or video.startswith("""https://"""		):
     _snake_case							:		Tuple								=  BytesIO(requests.get(a_		).content		)

   _snake_case							:		Union[str, Any]								=  VideoReader(a_		)
   videoreader.seek(0		)

   _snake_case							:		Union[str, Any]								=  0
   _snake_case							:		List[str]								=  num_frames * frame_sampling_rate - 1
   _snake_case							:		Dict								=  np.linspace(a_,      a_,      num=a_,      dtype=np.intaa		)

   _snake_case							:		str								=  videoreader.get_batch(a_		).asnumpy()
   _snake_case							:		Union[str, Any]								=  list(a_		)

   _snake_case							:		Union[str, Any]								=  self.image_processor(a_,      return_tensors=self.framework		)
   return model_inputs


 def  UpperCamelCase_   (				self:		str,      a_:		str		):




   '''simple docstring'''



   _snake_case							:		List[Any]								=  self.model(**a_		)
   return model_outputs


 def  UpperCamelCase_   (				self:		Union[str, Any],      a_:		List[str],      a_:		str=5		):




   '''simple docstring'''



   if top_k > self.model.config.num_labels:
     _snake_case							:		Optional[int]								=  self.model.config.num_labels

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

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








import torch

from diffusers import UnCLIPScheduler

from .test_schedulers import SchedulerCommonTest


class  A		(  __UpperCAmelCase     ):
		__snake_case									=   (UnCLIPScheduler,)



		def        SCREAMING_SNAKE_CASE__			(     self,	**UpperCamelCase__   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  {
								    '''num_train_timesteps''': 1000,
								    '''variance_type''': '''fixed_small_log''',
								    '''clip_sample''': True,
								    '''clip_sample_range''': 1.0,
								    '''prediction_type''': '''epsilon''',
								}

								config.update(**UpperCamelCase__   )
								return config



		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								for timesteps in [1, 5, 100, 1000]:
														self.check_over_configs(num_train_timesteps=UpperCamelCase__   )



		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								for variance in ["fixed_small_log", "learned_range"]:
														self.check_over_configs(variance_type=UpperCamelCase__   )



		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								for clip_sample in [True, False]:
														self.check_over_configs(clip_sample=UpperCamelCase__   )



		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								for clip_sample_range in [1, 5, 10, 20]:
														self.check_over_configs(clip_sample_range=UpperCamelCase__   )



		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								for prediction_type in ["epsilon", "sample"]:
														self.check_over_configs(prediction_type=UpperCamelCase__   )



		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								for time_step in [0, 500, 999]:
														for prev_timestep in [None, 5, 100, 250, 500, 750]:
																				if prev_timestep is not None and prev_timestep >= time_step:
																										continue

																				self.check_over_forward(time_step=UpperCamelCase__,	prev_timestep=UpperCamelCase__   )



		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  self.scheduler_classes[0]
								lowerCAmelCase_      			=  self.get_scheduler_config(variance_type='''fixed_small_log'''   )
								lowerCAmelCase_      			=  scheduler_class(**UpperCamelCase__   )

								assert torch.sum(torch.abs(scheduler._get_variance(0   ) - 1.0000E-10   )   ) < 1E-5
								assert torch.sum(torch.abs(scheduler._get_variance(487   ) - 0.0_549_625   )   ) < 1E-5
								assert torch.sum(torch.abs(scheduler._get_variance(999   ) - 0.9_994_987   )   ) < 1E-5



		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  self.scheduler_classes[0]
								lowerCAmelCase_      			=  self.get_scheduler_config(variance_type='''learned_range'''   )
								lowerCAmelCase_      			=  scheduler_class(**UpperCamelCase__   )

								lowerCAmelCase_      			=  0.5

								assert scheduler._get_variance(1,	predicted_variance=UpperCamelCase__   ) - -10.1_712_790 < 1E-5
								assert scheduler._get_variance(487,	predicted_variance=UpperCamelCase__   ) - -5.7_998_052 < 1E-5
								assert scheduler._get_variance(999,	predicted_variance=UpperCamelCase__   ) - -0.0_010_011 < 1E-5



		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  self.scheduler_classes[0]
								lowerCAmelCase_      			=  self.get_scheduler_config()
								lowerCAmelCase_      			=  scheduler_class(**UpperCamelCase__   )

								lowerCAmelCase_      			=  scheduler.timesteps

								lowerCAmelCase_      			=  self.dummy_model()
								lowerCAmelCase_      			=  self.dummy_sample_deter
								lowerCAmelCase_      			=  torch.manual_seed(0   )

								for i, t in enumerate(UpperCamelCase__   ):
														# 1. predict noise residual
														lowerCAmelCase_      			=  model(UpperCamelCase__,	UpperCamelCase__   )

														# 2. predict previous mean of sample x_t-1
														lowerCAmelCase_      			=  scheduler.step(UpperCamelCase__,	UpperCamelCase__,	UpperCamelCase__,	generator=UpperCamelCase__   ).prev_sample

														lowerCAmelCase_      			=  pred_prev_sample

								lowerCAmelCase_      			=  torch.sum(torch.abs(UpperCamelCase__   )   )
								lowerCAmelCase_      			=  torch.mean(torch.abs(UpperCamelCase__   )   )

								assert abs(result_sum.item() - 252.2_682_495   ) < 1E-2
								assert abs(result_mean.item() - 0.3_284_743   ) < 1E-3



		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								lowerCAmelCase_      			=  self.scheduler_classes[0]
								lowerCAmelCase_      			=  self.get_scheduler_config()
								lowerCAmelCase_      			=  scheduler_class(**UpperCamelCase__   )

								scheduler.set_timesteps(25   )

								lowerCAmelCase_      			=  scheduler.timesteps

								lowerCAmelCase_      			=  self.dummy_model()
								lowerCAmelCase_      			=  self.dummy_sample_deter
								lowerCAmelCase_      			=  torch.manual_seed(0   )

								for i, t in enumerate(UpperCamelCase__   ):
														# 1. predict noise residual
														lowerCAmelCase_      			=  model(UpperCamelCase__,	UpperCamelCase__   )

														if i + 1 == timesteps.shape[0]:
																				lowerCAmelCase_      			=  None
														else:
																				lowerCAmelCase_      			=  timesteps[i + 1]

														# 2. predict previous mean of sample x_t-1
														lowerCAmelCase_      			=  scheduler.step(
														    UpperCamelCase__,	UpperCamelCase__,	UpperCamelCase__,	prev_timestep=UpperCamelCase__,	generator=UpperCamelCase__   ).prev_sample

														lowerCAmelCase_      			=  pred_prev_sample

								lowerCAmelCase_      			=  torch.sum(torch.abs(UpperCamelCase__   )   )
								lowerCAmelCase_      			=  torch.mean(torch.abs(UpperCamelCase__   )   )

								assert abs(result_sum.item() - 258.2_044_983   ) < 1E-2
								assert abs(result_mean.item() - 0.3_362_038   ) < 1E-3



		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								pass







		def        SCREAMING_SNAKE_CASE__			(     self   ):







								"""simple docstring"""



								pass




