"""simple docstring"""




import os
import tempfile
import unittest
from pathlib import Path

from transformers import AutoConfig, is_torch_available
from transformers.testing_utils import require_torch, torch_device


if is_torch_available():
				from transformers import PyTorchBenchmark, PyTorchBenchmarkArguments
@require_torch
class 			UpperCamelCase__(	unittest.TestCase       ):





						def  snake_case__		(				self						,__UpperCAmelCase   )    -> Any:
													for model_result in results.values():
																				for batch_size, sequence_length in zip(model_result['bs']						,model_result['ss']   ):
																											A__		   =							model_result['result'][batch_size][sequence_length]
																											self.assertIsNotNone(__UpperCAmelCase   )





						def  snake_case__		(				self   )    -> Any:
													A__		   =							'sshleifer/tiny-gpt2'
													A__		   =							PyTorchBenchmarkArguments(
													    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,multi_process=__UpperCAmelCase						,)
													A__		   =							PyTorchBenchmark(__UpperCAmelCase   )
													A__		   =							benchmark.run()
													self.check_results_dict_not_empty(results.time_inference_result   )
													self.check_results_dict_not_empty(results.memory_inference_result   )





						def  snake_case__		(				self   )    -> Union[str, Any]:
													A__		   =							'sgugger/tiny-distilbert-classification'
													A__		   =							PyTorchBenchmarkArguments(
													    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,multi_process=__UpperCAmelCase						,only_pretrain_model=__UpperCAmelCase						,)
													A__		   =							PyTorchBenchmark(__UpperCAmelCase   )
													A__		   =							benchmark.run()
													self.check_results_dict_not_empty(results.time_inference_result   )
													self.check_results_dict_not_empty(results.memory_inference_result   )





						def  snake_case__		(				self   )    -> Union[str, Any]:
													A__		   =							'sshleifer/tiny-gpt2'
													A__		   =							PyTorchBenchmarkArguments(
													    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,torchscript=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,multi_process=__UpperCAmelCase						,)
													A__		   =							PyTorchBenchmark(__UpperCAmelCase   )
													A__		   =							benchmark.run()
													self.check_results_dict_not_empty(results.time_inference_result   )
													self.check_results_dict_not_empty(results.memory_inference_result   )





						@unittest.skipIf(torch_device == 'cpu'						,'Cant do half precision'   )
						def  snake_case__		(				self   )    -> str:
													A__		   =							'sshleifer/tiny-gpt2'
													A__		   =							PyTorchBenchmarkArguments(
													    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,fpaa=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,multi_process=__UpperCAmelCase						,)
													A__		   =							PyTorchBenchmark(__UpperCAmelCase   )
													A__		   =							benchmark.run()
													self.check_results_dict_not_empty(results.time_inference_result   )
													self.check_results_dict_not_empty(results.memory_inference_result   )





						def  snake_case__		(				self   )    -> Optional[Any]:
													A__		   =							'sshleifer/tiny-gpt2'
													A__		   =							AutoConfig.from_pretrained(__UpperCAmelCase   )
													# set architectures equal to `None`
													A__		   =							None
													A__		   =							PyTorchBenchmarkArguments(
													    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,multi_process=__UpperCAmelCase						,)
													A__		   =							PyTorchBenchmark(__UpperCAmelCase						,configs=[config]   )
													A__		   =							benchmark.run()
													self.check_results_dict_not_empty(results.time_inference_result   )
													self.check_results_dict_not_empty(results.memory_inference_result   )





						def  snake_case__		(				self   )    -> List[Any]:
													A__		   =							'sshleifer/tiny-gpt2'
													A__		   =							PyTorchBenchmarkArguments(
													    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,multi_process=__UpperCAmelCase						,)
													A__		   =							PyTorchBenchmark(__UpperCAmelCase   )
													A__		   =							benchmark.run()
													self.check_results_dict_not_empty(results.time_train_result   )
													self.check_results_dict_not_empty(results.memory_train_result   )





						@unittest.skipIf(torch_device == 'cpu'						,'Can\'t do half precision'   )
						def  snake_case__		(				self   )    -> int:
													A__		   =							'sshleifer/tiny-gpt2'
													A__		   =							PyTorchBenchmarkArguments(
													    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,fpaa=__UpperCAmelCase						,multi_process=__UpperCAmelCase						,)
													A__		   =							PyTorchBenchmark(__UpperCAmelCase   )
													A__		   =							benchmark.run()
													self.check_results_dict_not_empty(results.time_train_result   )
													self.check_results_dict_not_empty(results.memory_train_result   )





						def  snake_case__		(				self   )    -> List[str]:
													A__		   =							'sshleifer/tiny-gpt2'
													A__		   =							AutoConfig.from_pretrained(__UpperCAmelCase   )
													A__		   =							PyTorchBenchmarkArguments(
													    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,multi_process=__UpperCAmelCase						,)
													A__		   =							PyTorchBenchmark(__UpperCAmelCase						,configs=[config]   )
													A__		   =							benchmark.run()
													self.check_results_dict_not_empty(results.time_inference_result   )
													self.check_results_dict_not_empty(results.memory_inference_result   )





						def  snake_case__		(				self   )    -> Optional[Any]:
													A__		   =							'sshleifer/tinier_bart'
													A__		   =							AutoConfig.from_pretrained(__UpperCAmelCase   )
													A__		   =							PyTorchBenchmarkArguments(
													    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,multi_process=__UpperCAmelCase						,)
													A__		   =							PyTorchBenchmark(__UpperCAmelCase						,configs=[config]   )
													A__		   =							benchmark.run()
													self.check_results_dict_not_empty(results.time_inference_result   )
													self.check_results_dict_not_empty(results.memory_inference_result   )





						def  snake_case__		(				self   )    -> str:
													A__		   =							'sshleifer/tiny-gpt2'
													A__		   =							AutoConfig.from_pretrained(__UpperCAmelCase   )
													A__		   =							PyTorchBenchmarkArguments(
													    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,multi_process=__UpperCAmelCase						,)
													A__		   =							PyTorchBenchmark(__UpperCAmelCase						,configs=[config]   )
													A__		   =							benchmark.run()
													self.check_results_dict_not_empty(results.time_train_result   )
													self.check_results_dict_not_empty(results.memory_train_result   )





						def  snake_case__		(				self   )    -> Optional[Any]:
													A__		   =							'sshleifer/tinier_bart'
													A__		   =							AutoConfig.from_pretrained(__UpperCAmelCase   )
													A__		   =							PyTorchBenchmarkArguments(
													    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,multi_process=__UpperCAmelCase						,)
													A__		   =							PyTorchBenchmark(__UpperCAmelCase						,configs=[config]   )
													A__		   =							benchmark.run()
													self.check_results_dict_not_empty(results.time_train_result   )
													self.check_results_dict_not_empty(results.memory_train_result   )





						def  snake_case__		(				self   )    -> List[str]:
													A__		   =							'sshleifer/tiny-gpt2'
													with tempfile.TemporaryDirectory() as tmp_dir:
																				A__		   =							PyTorchBenchmarkArguments(
																				    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,save_to_csv=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,inference_time_csv_file=os.path.join(__UpperCAmelCase						,'inf_time.csv'   )						,train_memory_csv_file=os.path.join(__UpperCAmelCase						,'train_mem.csv'   )						,inference_memory_csv_file=os.path.join(__UpperCAmelCase						,'inf_mem.csv'   )						,train_time_csv_file=os.path.join(__UpperCAmelCase						,'train_time.csv'   )						,env_info_csv_file=os.path.join(__UpperCAmelCase						,'env.csv'   )						,multi_process=__UpperCAmelCase						,)
																				A__		   =							PyTorchBenchmark(__UpperCAmelCase   )
																				benchmark.run()
																				self.assertTrue(Path(os.path.join(__UpperCAmelCase						,'inf_time.csv'   )   ).exists()   )
																				self.assertTrue(Path(os.path.join(__UpperCAmelCase						,'train_time.csv'   )   ).exists()   )
																				self.assertTrue(Path(os.path.join(__UpperCAmelCase						,'inf_mem.csv'   )   ).exists()   )
																				self.assertTrue(Path(os.path.join(__UpperCAmelCase						,'train_mem.csv'   )   ).exists()   )
																				self.assertTrue(Path(os.path.join(__UpperCAmelCase						,'env.csv'   )   ).exists()   )





						def  snake_case__		(				self   )    -> List[Any]:
													A__		   =							'sshleifer/tiny-gpt2'

													def _check_summary_is_not_empty(__UpperCAmelCase   ):
																				self.assertTrue(hasattr(__UpperCAmelCase						,'sequential'   )   )
																				self.assertTrue(hasattr(__UpperCAmelCase						,'cumulative'   )   )
																				self.assertTrue(hasattr(__UpperCAmelCase						,'current'   )   )
																				self.assertTrue(hasattr(__UpperCAmelCase						,'total'   )   )

													with tempfile.TemporaryDirectory() as tmp_dir:
																				A__		   =							PyTorchBenchmarkArguments(
																				    models=[MODEL_ID]						,training=__UpperCAmelCase						,inference=__UpperCAmelCase						,sequence_lengths=[8]						,batch_sizes=[1]						,log_filename=os.path.join(__UpperCAmelCase						,'log.txt'   )						,log_print=__UpperCAmelCase						,trace_memory_line_by_line=__UpperCAmelCase						,multi_process=__UpperCAmelCase						,)
																				A__		   =							PyTorchBenchmark(__UpperCAmelCase   )
																				A__		   =							benchmark.run()
																				_check_summary_is_not_empty(result.inference_summary   )
																				_check_summary_is_not_empty(result.train_summary   )
																				self.assertTrue(Path(os.path.join(__UpperCAmelCase						,'log.txt'   )   ).exists()   )






"""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(__A       )
class 			UpperCamelCase__(	__A       ):





						def __init__(				self						,*__UpperCAmelCase						,**__UpperCAmelCase   )    -> Tuple:
													super().__init__(*__UpperCAmelCase						,**__UpperCAmelCase   )
													requires_backends(self						,'decord'   )
													self.check_model_type(__UpperCAmelCase   )





						def  snake_case__		(				self						,__UpperCAmelCase=None						,__UpperCAmelCase=None						,__UpperCAmelCase=None   )    -> int:
													A__		   =							{}
													if frame_sampling_rate is not None:
																				A__		   =							frame_sampling_rate
													if num_frames is not None:
																				A__		   =							num_frames

													A__		   =							{}
													if top_k is not None:
																				A__		   =							top_k
													return preprocess_params, {}, postprocess_params





						def __call__(				self						,__UpperCAmelCase						,**__UpperCAmelCase   )    -> Dict:
													return super().__call__(__UpperCAmelCase						,**__UpperCAmelCase   )





						def  snake_case__		(				self						,__UpperCAmelCase						,__UpperCAmelCase=None						,__UpperCAmelCase=1   )    -> Union[str, Any]:
													if num_frames is None:
																				A__		   =							self.model.config.num_frames

													if video.startswith('http://'   ) or video.startswith('https://'   ):
																				A__		   =							BytesIO(requests.get(__UpperCAmelCase   ).content   )

													A__		   =							VideoReader(__UpperCAmelCase   )
													videoreader.seek(0   )

													A__		   =							0
													A__		   =							num_frames * frame_sampling_rate - 1
													A__		   =							np.linspace(__UpperCAmelCase						,__UpperCAmelCase						,num=__UpperCAmelCase						,dtype=np.intaa   )

													A__		   =							videoreader.get_batch(__UpperCAmelCase   ).asnumpy()
													A__		   =							list(__UpperCAmelCase   )

													A__		   =							self.image_processor(__UpperCAmelCase						,return_tensors=self.framework   )
													return model_inputs





						def  snake_case__		(				self						,__UpperCAmelCase   )    -> Dict:
													A__		   =							self.model(**__UpperCAmelCase   )
													return model_outputs





						def  snake_case__		(				self						,__UpperCAmelCase						,__UpperCAmelCase=5   )    -> Union[str, Any]:
													if top_k > self.model.config.num_labels:
																				A__		   =							self.model.config.num_labels

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

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





