






'''simple docstring'''

import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import TimesformerConfig
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
	import torch
	from torch import nn

	from transformers import (
	    MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING,
	    TimesformerForVideoClassification,
	    TimesformerModel,
	)
	from transformers.models.timesformer.modeling_timesformer import TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
	from transformers import VideoMAEImageProcessor






class 							__magic_name__     :







	"""simple docstring"""



	def __init__(							self  ,      _lowercase  ,      _lowercase=13  ,      _lowercase=10  ,      _lowercase=3  ,      _lowercase=2  ,      _lowercase=2  ,      _lowercase=True  ,      _lowercase=True  ,      _lowercase=32  ,      _lowercase=5  ,      _lowercase=4  ,      _lowercase=37  ,      _lowercase="gelu"  ,      _lowercase=0.1  ,      _lowercase=0.1  ,      _lowercase=10  ,      _lowercase=0.02  ,      _lowercase="divided_space_time"  ,      _lowercase=None  ,      )						->  Tuple:
							lowercase_      :   List[str]        =			parent
							lowercase_      :   str        =			batch_size
							lowercase_      :   Optional[Any]        =			image_size
							lowercase_      :   Dict        =			num_channels
							lowercase_      :   List[str]        =			patch_size
							lowercase_      :   int        =			num_frames
							lowercase_      :   List[str]        =			is_training
							lowercase_      :   Union[str, Any]        =			use_labels
							lowercase_      :   int        =			hidden_size
							lowercase_      :   List[Any]        =			num_hidden_layers
							lowercase_      :   str        =			num_attention_heads
							lowercase_      :   int        =			intermediate_size
							lowercase_      :   Tuple        =			hidden_act
							lowercase_      :   List[str]        =			hidden_dropout_prob
							lowercase_      :   int        =			attention_probs_dropout_prob
							lowercase_      :   List[str]        =			attention_type
							lowercase_      :   int        =			initializer_range
							lowercase_      :   List[str]        =			scope
							lowercase_      :   Any        =			num_labels

							# in TimeSformer, the number of spatial tokens equals num_frames * num_patches per frame + 1 CLS token
							lowercase_      :   List[Any]        =			(image_size // patch_size) ** 2
							lowercase_      :   str        =			(num_frames) * self.num_patches_per_frame + 1
	def        lowerCamelCase__		(							self       )						->  int:
							lowercase_      :   Optional[int]        =			floats_tensor(
							    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]       )

							lowercase_      :   Any        =			None
							if self.use_labels:
													lowercase_      :   str        =			ids_tensor([self.batch_size]  ,      self.num_labels       )

							lowercase_      :   Optional[Any]        =			self.get_config()

							return config, pixel_values, labels
	def        lowerCamelCase__		(							self       )						->  List[str]:
							lowercase_      :   Tuple        =			TimesformerConfig(
							    image_size=self.image_size  ,      patch_size=self.patch_size  ,      num_channels=self.num_channels  ,      num_frames=self.num_frames  ,      hidden_size=self.hidden_size  ,      num_hidden_layers=self.num_hidden_layers  ,      num_attention_heads=self.num_attention_heads  ,      intermediate_size=self.intermediate_size  ,      hidden_act=self.hidden_act  ,      hidden_dropout_prob=self.hidden_dropout_prob  ,      attention_probs_dropout_prob=self.attention_probs_dropout_prob  ,      initializer_range=self.initializer_range  ,      attention_type=self.attention_type  ,      )
							lowercase_      :   List[Any]        =			self.num_labels
							return config
	def        lowerCamelCase__		(							self  ,      _lowercase  ,      _lowercase  ,      _lowercase       )						->  List[str]:
							lowercase_      :   Any        =			TimesformerModel(config=__A       )
							model.to(__A       )
							model.eval()
							lowercase_      :   int        =			model(__A       )
							self.parent.assertEqual(result.last_hidden_state.shape  ,      (self.batch_size, self.seq_length, self.hidden_size)       )
	def        lowerCamelCase__		(							self  ,      _lowercase  ,      _lowercase  ,      _lowercase       )						->  str:
							lowercase_      :   Any        =			TimesformerForVideoClassification(__A       )
							model.to(__A       )
							model.eval()

							lowercase_      :   Dict        =			model(__A       )

							# verify the logits shape
							lowercase_      :   Union[str, Any]        =			torch.Size((self.batch_size, self.num_labels)       )
							self.parent.assertEqual(result.logits.shape  ,      __A       )







	def        lowerCamelCase__		(							self       )						->  List[str]:
							lowercase_      :   Dict        =			self.prepare_config_and_inputs()
							lowercase_      :   List[str]        =			config_and_inputs
							lowercase_      :   Tuple        =			{"pixel_values": pixel_values}
							return config, inputs_dict






@require_torch
class 							__magic_name__     (	UpperCamelCase_,						UpperCamelCase_,						unittest.TestCase				):







	"""simple docstring"""



	SCREAMING_SNAKE_CASE_					:      Union[str, Any]        =  (TimesformerModel, TimesformerForVideoClassification) if is_torch_available() else ()
	SCREAMING_SNAKE_CASE_					:      Dict        =  (
	    {'feature-extraction': TimesformerModel, 'video-classification': TimesformerForVideoClassification}
	    if is_torch_available()
	    else {}
	)

	SCREAMING_SNAKE_CASE_					:      int        =  False
	SCREAMING_SNAKE_CASE_					:      Optional[int]        =  False
	SCREAMING_SNAKE_CASE_					:      int        =  False
	SCREAMING_SNAKE_CASE_					:      Union[str, Any]        =  False
	def        lowerCamelCase__		(							self       )						->  Tuple:
							lowercase_      :   Optional[int]        =			TimesformerModelTester(self       )
							lowercase_      :   Tuple        =			ConfigTester(
							    self  ,      config_class=__A  ,      has_text_modality=__A  ,      hidden_size=37       )
	def        lowerCamelCase__		(							self  ,      _lowercase  ,      _lowercase  ,      _lowercase=False       )						->  Dict:
							lowercase_      :   Optional[Any]        =			copy.deepcopy(__A       )

							if return_labels:
													if model_class in get_values(__A       ):
																			lowercase_      :   List[str]        =			torch.zeros(
																			    self.model_tester.batch_size  ,      dtype=torch.long  ,      device=__A       )

							return inputs_dict
	def        lowerCamelCase__		(							self       )						->  List[Any]:
							self.config_tester.run_common_tests()
	@unittest.skip(reason='TimeSformer does not use inputs_embeds'       )
	def        lowerCamelCase__		(							self       )						->  List[Any]:
							pass
	def        lowerCamelCase__		(							self       )						->  Dict:
							lowercase_      :   Union[str, Any]        =			self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
													lowercase_      :   Any        =			model_class(__A       )
													self.assertIsInstance(model.get_input_embeddings()  ,      (nn.Module)       )
													lowercase_      :   Tuple        =			model.get_output_embeddings()
													self.assertTrue(x is None or isinstance(__A  ,      nn.Linear       )       )
	def        lowerCamelCase__		(							self       )						->  Any:
							lowercase_      :   str        =			self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
													lowercase_      :   Any        =			model_class(__A       )
													lowercase_      :   Tuple        =			inspect.signature(model.forward       )
													# signature.parameters is an OrderedDict => so arg_names order is deterministic
													lowercase_      :   Union[str, Any]        =			[*signature.parameters.keys()]

													lowercase_      :   int        =			["pixel_values"]
													self.assertListEqual(arg_names[:1]  ,      __A       )
	def        lowerCamelCase__		(							self       )						->  List[Any]:
							lowercase_      :   Optional[Any]        =			self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_model(*__A       )
	def        lowerCamelCase__		(							self       )						->  List[Any]:
							lowercase_      :   str        =			self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_video_classification(*__A       )
	@slow
	def        lowerCamelCase__		(							self       )						->  Union[str, Any]:
							for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
													lowercase_      :   List[str]        =			TimesformerModel.from_pretrained(__A       )
													self.assertIsNotNone(__A       )
	def        lowerCamelCase__		(							self       )						->  List[Any]:
							if not self.has_attentions:
													pass

							else:
													lowercase_      :   int        =			self.model_tester.prepare_config_and_inputs_for_common()
													lowercase_      :   Union[str, Any]        =			True

													for model_class in self.all_model_classes:
																			lowercase_      :   List[Any]        =			self.model_tester.seq_length
																			lowercase_      :   Optional[int]        =			self.model_tester.num_frames

																			lowercase_      :   int        =			True
																			lowercase_      :   Optional[Any]        =			False
																			lowercase_      :   Optional[int]        =			True
																			lowercase_      :   Optional[int]        =			model_class(__A       )
																			model.to(__A       )
																			model.eval()
																			with torch.no_grad():
																									lowercase_      :   str        =			model(**self._prepare_for_class(__A  ,      __A       )       )
																			lowercase_      :   Dict        =			outputs.attentions
																			self.assertEqual(len(__A       )  ,      self.model_tester.num_hidden_layers       )

																			# check that output_attentions also work using config
																			del inputs_dict["output_attentions"]
																			lowercase_      :   List[Any]        =			True
																			lowercase_      :   List[str]        =			model_class(__A       )
																			model.to(__A       )
																			model.eval()
																			with torch.no_grad():
																									lowercase_      :   int        =			model(**self._prepare_for_class(__A  ,      __A       )       )
																			lowercase_      :   Dict        =			outputs.attentions
																			self.assertEqual(len(__A       )  ,      self.model_tester.num_hidden_layers       )

																			# attentions has shape (batch_size x num_frames) x num_heads x (num_patches per frame + 1) x (num_patches per frame + 1)
																			self.assertListEqual(
																			    list(attentions[0].shape[-3:]       )  ,      [self.model_tester.num_attention_heads, seq_len // num_frames + 1, seq_len // num_frames + 1]  ,      )
																			lowercase_      :   Dict        =			len(__A       )

																			# Check attention is always last and order is fine
																			lowercase_      :   Union[str, Any]        =			True
																			lowercase_      :   List[str]        =			True
																			lowercase_      :   Optional[Any]        =			model_class(__A       )
																			model.to(__A       )
																			model.eval()
																			with torch.no_grad():
																									lowercase_      :   str        =			model(**self._prepare_for_class(__A  ,      __A       )       )

																			self.assertEqual(out_len + 1  ,      len(__A       )       )

																			lowercase_      :   str        =			outputs.attentions

																			self.assertEqual(len(__A       )  ,      self.model_tester.num_hidden_layers       )

																			# attentions has shape (batch_size x num_frames) x num_heads x (num_patches per frame + 1) x (num_patches per frame + 1)
																			self.assertListEqual(
																			    list(self_attentions[0].shape[-3:]       )  ,      [self.model_tester.num_attention_heads, seq_len // num_frames + 1, seq_len // num_frames + 1]  ,      )







	def        lowerCamelCase__		(							self       )						->  Optional[Any]:
							def check_hidden_states_output(_lowercase  ,      _lowercase  ,      _lowercase       ):
													lowercase_      :   Optional[int]        =			model_class(__A       )
													model.to(__A       )
													model.eval()

													with torch.no_grad():
																			lowercase_      :   str        =			model(**self._prepare_for_class(__A  ,      __A       )       )

													lowercase_      :   Dict        =			outputs.hidden_states
													lowercase_      :   Union[str, Any]        =			self.model_tester.num_hidden_layers + 1
													self.assertEqual(len(__A       )  ,      __A       )

													lowercase_      :   Tuple        =			self.model_tester.seq_length

													self.assertListEqual(
													    list(hidden_states[0].shape[-2:]       )  ,      [seq_length, self.model_tester.hidden_size]  ,      )

							lowercase_      :   int        =			self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
													lowercase_      :   int        =			True
													check_hidden_states_output(__A  ,      __A  ,      __A       )

													# check that output_hidden_states also work using config
													del inputs_dict["output_hidden_states"]
													lowercase_      :   str        =			True

													check_hidden_states_output(__A  ,      __A  ,      __A       )






def 	_UpperCAmelCase    (  )			->			Optional[int]:





						"""simple docstring"""
						lowercase_      :   List[Any]        =			hf_hub_download(
						    repo_id='hf-internal-testing/spaghetti-video'						,			filename='eating_spaghetti.npy'						,			repo_type='dataset'							)
						lowercase_      :   List[Any]        =			np.load(snake_case_							)
						return list(snake_case_							)






@require_torch
@require_vision
class 							__magic_name__     (	unittest.TestCase				):







	"""simple docstring"""



	@cached_property
	def        lowerCamelCase__		(							self       )						->  List[Any]:
							# logits were tested with a different mean and std, so we use the same here
							return (
							    VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]  ,      image_std=[0.5, 0.5, 0.5]       )
							    if is_vision_available()
							    else None
							)







	@slow
	def        lowerCamelCase__		(							self       )						->  Optional[Any]:
							lowercase_      :   Tuple        =			TimesformerForVideoClassification.from_pretrained('facebook/timesformer-base-finetuned-k400'       ).to(
							    __A       )

							lowercase_      :   Tuple        =			self.default_image_processor
							lowercase_      :   str        =			prepare_video()
							lowercase_      :   str        =			image_processor(video[:8]  ,      return_tensors='pt'       ).to(__A       )

							# forward pass
							with torch.no_grad():
													lowercase_      :   Optional[Any]        =			model(**__A       )

							# verify the logits
							lowercase_      :   Optional[Any]        =			torch.Size((1, 400)       )
							self.assertEqual(outputs.logits.shape  ,      __A       )

							lowercase_      :   Tuple        =			torch.tensor([-0.30_16, -0.77_13, -0.42_05]       ).to(__A       )

							self.assertTrue(torch.allclose(outputs.logits[0, :3]  ,      __A  ,      atol=1E-4       )       )








'''simple docstring'''

class 							__magic_name__     :







	"""simple docstring"""



	def __init__(							self  ,      _lowercase       )						->  Union[str, Any]:
							lowercase_      :   Dict        =			n
							lowercase_      :   Dict        =			[None] * self.n
							lowercase_      :   Tuple        =			0  # index of the first element
							lowercase_      :   List[Any]        =			0
							lowercase_      :   List[Any]        =			0
	def __len__(							self       )						->  int:
							return self.size
	def        lowerCamelCase__		(							self       )						->  bool:
							return self.size == 0
	def        lowerCamelCase__		(							self       )						->  List[Any]:
							return False if self.is_empty() else self.array[self.front]
	def        lowerCamelCase__		(							self  ,      _lowercase       )						->  Any:
							if self.size >= self.n:
													raise Exception('QUEUE IS FULL'       )

							lowercase_      :   Tuple        =			data
							lowercase_      :   List[Any]        =			(self.rear + 1) % self.n
							self.size += 1
							return self







	def        lowerCamelCase__		(							self       )						->  Any:
							if self.size == 0:
													raise Exception('UNDERFLOW'       )

							lowercase_      :   Dict        =			self.array[self.front]
							lowercase_      :   Tuple        =			None
							lowercase_      :   int        =			(self.front + 1) % self.n
							self.size -= 1
							return temp
