

"""simple docstring"""




import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import VideoMAEConfig
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,
				    VideoMAEForPreTraining,
				    VideoMAEForVideoClassification,
				    VideoMAEModel,
				)
				from transformers.models.videomae.modeling_videomae import VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
				from transformers import VideoMAEImageProcessor
class 			_UpperCAmelCase					:



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

		# in VideoMAE, the number of tokens equals num_frames/tubelet_size * num_patches per frame
		__UpperCAmelCase							=      (image_size // patch_size) ** 2
		__UpperCAmelCase							=      (num_frames // tubelet_size) * self.num_patches_per_frame

		# use this variable to define bool_masked_pos
		__UpperCAmelCase							=      int(mask_ratio * self.seq_length		)



	def     a   (   self			:    Any		):
		__UpperCAmelCase							=      floats_tensor(
		    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]		)

		__UpperCAmelCase							=      None
		if self.use_labels:
			__UpperCAmelCase							=      ids_tensor([self.batch_size]       ,  self.type_sequence_label_size		)

		__UpperCAmelCase							=      self.get_config()

		return config, pixel_values, labels



	def     a   (   self			:    List[str]		):
		return VideoMAEConfig(
		    image_size=self.image_size       ,  patch_size=self.patch_size       ,  num_channels=self.num_channels       ,  num_frames=self.num_frames       ,  tubelet_size=self.tubelet_size       ,  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       ,  is_decoder=lowercase__       ,  initializer_range=self.initializer_range       ,  )



	def     a   (   self			:    int       ,  _lowercase			:    int       ,  _lowercase			:    Any       ,  _lowercase			:    Optional[Any]		):
		__UpperCAmelCase							=      VideoMAEModel(config=lowercase__		)
		model.to(lowercase__		)
		model.eval()
		__UpperCAmelCase							=      model(lowercase__		)
		self.parent.assertEqual(result.last_hidden_state.shape       ,  (self.batch_size, self.seq_length, self.hidden_size)		)



	def     a   (   self			:    Dict       ,  _lowercase			:    Optional[Any]       ,  _lowercase			:    Any       ,  _lowercase			:    Any		):
		__UpperCAmelCase							=      VideoMAEForPreTraining(lowercase__		)
		model.to(lowercase__		)
		model.eval()
		# important: each video needs to have the same number of masked patches
		# hence we define a single mask, which we then repeat for each example in the batch
		__UpperCAmelCase							=      torch.ones((self.num_masks,)		)
		__UpperCAmelCase							=      torch.cat([mask, torch.zeros(self.seq_length - mask.size(0		)		)]		)
		__UpperCAmelCase							=      mask.expand(self.batch_size       ,  -1		).bool()

		__UpperCAmelCase							=      model(lowercase__       ,  lowercase__		)
		# model only returns predictions for masked patches
		__UpperCAmelCase							=      mask.sum().item()
		__UpperCAmelCase							=      3 * self.tubelet_size * self.patch_size**2
		self.parent.assertEqual(result.logits.shape       ,  (self.batch_size, num_masked_patches, decoder_num_labels)		)



	def     a   (   self			:    Tuple		):
		__UpperCAmelCase							=      self.prepare_config_and_inputs()
		__UpperCAmelCase						,				__UpperCAmelCase						,				__UpperCAmelCase							=      config_and_inputs
		__UpperCAmelCase							=      {'''pixel_values''': pixel_values}
		return config, inputs_dict






@require_torch
class 			_UpperCAmelCase					(				lowerCamelCase__						,  lowerCamelCase__						,  unittest.TestCase    ):
	a__						:      Optional[Any]						       = (
	    (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
	)
	a__						:      int						       = (
	    {'feature-extraction': VideoMAEModel, 'video-classification': VideoMAEForVideoClassification}
	    if is_torch_available()
	    else {}
	)

	a__						:      str						       = False
	a__						:      Any						       = False
	a__						:      str						       = False
	a__						:      Union[str, Any]						       = False



	def     a   (   self			:    Any		):
		__UpperCAmelCase							=      VideoMAEModelTester(self		)
		__UpperCAmelCase							=      ConfigTester(self       ,  config_class=lowercase__       ,  has_text_modality=lowercase__       ,  hidden_size=37		)



	def     a   (   self			:    int       ,  _lowercase			:    Tuple       ,  _lowercase			:    List[str]       ,  _lowercase			:    List[str]=False		):
		__UpperCAmelCase							=      copy.deepcopy(lowercase__		)

		if model_class == VideoMAEForPreTraining:
			# important: each video needs to have the same number of masked patches
			# hence we define a single mask, which we then repeat for each example in the batch
			__UpperCAmelCase							=      torch.ones((self.model_tester.num_masks,)		)
			__UpperCAmelCase							=      torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0		)		)]		)
			__UpperCAmelCase							=      mask.expand(self.model_tester.batch_size       ,  -1		).bool()
			__UpperCAmelCase							=      bool_masked_pos.to(lowercase__		)

		if return_labels:
			if model_class in [
			    *get_values(lowercase__		),
			]:
				__UpperCAmelCase							=      torch.zeros(
				    self.model_tester.batch_size       ,  dtype=torch.long       ,  device=lowercase__		)

		return inputs_dict



	def     a   (   self			:    str		):
		self.config_tester.run_common_tests()



	@unittest.skip(reason='''VideoMAE does not use inputs_embeds'''		)
	def     a   (   self			:    Optional[Any]		):
		pass



	def     a   (   self			:    str		):
		__UpperCAmelCase						,				__UpperCAmelCase							=      self.model_tester.prepare_config_and_inputs_for_common()

		for model_class in self.all_model_classes:
			__UpperCAmelCase							=      model_class(lowercase__		)
			self.assertIsInstance(model.get_input_embeddings()       ,  (nn.Module)		)
			__UpperCAmelCase							=      model.get_output_embeddings()
			self.assertTrue(x is None or isinstance(lowercase__       ,  nn.Linear		)		)



	def     a   (   self			:    Optional[int]		):
		__UpperCAmelCase						,				__UpperCAmelCase							=      self.model_tester.prepare_config_and_inputs_for_common()

		for model_class in self.all_model_classes:
			__UpperCAmelCase							=      model_class(lowercase__		)
			__UpperCAmelCase							=      inspect.signature(model.forward		)
			# signature.parameters is an OrderedDict => so arg_names order is deterministic
			__UpperCAmelCase							=      [*signature.parameters.keys()]

			__UpperCAmelCase							=      ['''pixel_values''']
			self.assertListEqual(arg_names[:1]       ,  lowercase__		)



	def     a   (   self			:    Optional[Any]		):
		__UpperCAmelCase							=      self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_model(*lowercase__		)



	def     a   (   self			:    str		):
		__UpperCAmelCase							=      self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_for_pretraining(*lowercase__		)



	@slow
	def     a   (   self			:    Any		):
		for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
			__UpperCAmelCase							=      VideoMAEModel.from_pretrained(lowercase__		)
			self.assertIsNotNone(lowercase__		)



	def     a   (   self			:    Optional[int]		):
		if not self.has_attentions:
			pass

		else:
			__UpperCAmelCase						,				__UpperCAmelCase							=      self.model_tester.prepare_config_and_inputs_for_common()
			__UpperCAmelCase							=      True

			for model_class in self.all_model_classes:
				__UpperCAmelCase							=      self.model_tester.seq_length - self.model_tester.num_masks
				__UpperCAmelCase							=      (
				    num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length
				)

				__UpperCAmelCase							=      True
				__UpperCAmelCase							=      False
				__UpperCAmelCase							=      True
				__UpperCAmelCase							=      model_class(lowercase__		)
				model.to(lowercase__		)
				model.eval()
				with torch.no_grad():
					__UpperCAmelCase							=      model(**self._prepare_for_class(lowercase__       ,  lowercase__		)		)
				__UpperCAmelCase							=      outputs.attentions
				self.assertEqual(len(lowercase__		)       ,  self.model_tester.num_hidden_layers		)

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

				self.assertListEqual(
				    list(attentions[0].shape[-3:]		)       ,  [self.model_tester.num_attention_heads, seq_len, seq_len]       ,  )
				__UpperCAmelCase							=      len(lowercase__		)

				# Check attention is always last and order is fine
				__UpperCAmelCase							=      True
				__UpperCAmelCase							=      True
				__UpperCAmelCase							=      model_class(lowercase__		)
				model.to(lowercase__		)
				model.eval()
				with torch.no_grad():
					__UpperCAmelCase							=      model(**self._prepare_for_class(lowercase__       ,  lowercase__		)		)

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

				__UpperCAmelCase							=      outputs.attentions

				self.assertEqual(len(lowercase__		)       ,  self.model_tester.num_hidden_layers		)
				self.assertListEqual(
				    list(self_attentions[0].shape[-3:]		)       ,  [self.model_tester.num_attention_heads, seq_len, seq_len]       ,  )



	def     a   (   self			:    Dict		):
		def check_hidden_states_output(_lowercase			:    List[str]       ,  _lowercase			:    int       ,  _lowercase			:    int		):
			__UpperCAmelCase							=      model_class(lowercase__		)
			model.to(lowercase__		)
			model.eval()

			with torch.no_grad():
				__UpperCAmelCase							=      model(**self._prepare_for_class(lowercase__       ,  lowercase__		)		)

			__UpperCAmelCase							=      outputs.hidden_states
			__UpperCAmelCase							=      self.model_tester.num_hidden_layers + 1
			self.assertEqual(len(lowercase__		)       ,  lowercase__		)

			__UpperCAmelCase							=      self.model_tester.seq_length - self.model_tester.num_masks
			__UpperCAmelCase							=      num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length

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

		__UpperCAmelCase						,				__UpperCAmelCase							=      self.model_tester.prepare_config_and_inputs_for_common()

		for model_class in self.all_model_classes:
			__UpperCAmelCase							=      True
			check_hidden_states_output(lowercase__       ,  lowercase__       ,  lowercase__		)

			# check that output_hidden_states also work using config
			del inputs_dict["output_hidden_states"]
			__UpperCAmelCase							=      True

			check_hidden_states_output(lowercase__       ,  lowercase__       ,  lowercase__		)



	@unittest.skip('''Will be fixed soon by reducing the size of the model used for common tests.'''		)
	def     a   (   self			:    Tuple		):
		pass






def 	lowercase__		(    ):
	__UpperCAmelCase							=      hf_hub_download(
	    repo_id='''hf-internal-testing/spaghetti-video'''       ,						filename='''eating_spaghetti.npy'''       ,						repo_type='''dataset'''  )
	__UpperCAmelCase							=      np.load(A__  )
	return list(A__  )




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



	@cached_property
	def     a   (   self			:    List[str]		):
		# 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     a   (   self			:    Optional[Any]		):
		__UpperCAmelCase							=      VideoMAEForVideoClassification.from_pretrained('''MCG-NJU/videomae-base-finetuned-kinetics'''		).to(
		    lowercase__		)

		__UpperCAmelCase							=      self.default_image_processor
		__UpperCAmelCase							=      prepare_video()
		__UpperCAmelCase							=      image_processor(lowercase__       ,  return_tensors='''pt'''		).to(lowercase__		)

		# forward pass
		with torch.no_grad():
			__UpperCAmelCase							=      model(**lowercase__		)

		# verify the logits
		__UpperCAmelCase							=      torch.Size((1, 4_00)		)
		self.assertEqual(outputs.logits.shape       ,  lowercase__		)

		__UpperCAmelCase							=      torch.tensor([0.3_669, -0.0_688, -0.2_421]		).to(lowercase__		)

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



	@slow
	def     a   (   self			:    Tuple		):
		__UpperCAmelCase							=      VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''		).to(lowercase__		)

		__UpperCAmelCase							=      self.default_image_processor
		__UpperCAmelCase							=      prepare_video()
		__UpperCAmelCase							=      image_processor(lowercase__       ,  return_tensors='''pt'''		).to(lowercase__		)

		# add boolean mask, indicating which patches to mask
		__UpperCAmelCase							=      hf_hub_download(repo_id='''hf-internal-testing/bool-masked-pos'''       ,  filename='''bool_masked_pos.pt'''		)
		__UpperCAmelCase							=      torch.load(lowercase__		)

		# forward pass
		with torch.no_grad():
			__UpperCAmelCase							=      model(**lowercase__		)

		# verify the logits
		__UpperCAmelCase							=      torch.Size([1, 14_08, 15_36]		)
		__UpperCAmelCase							=      torch.tensor(
		    [[0.7_994, 0.9_612, 0.8_508], [0.7_401, 0.8_958, 0.8_302], [0.5_862, 0.7_468, 0.7_325]]       ,  device=lowercase__		)
		self.assertEqual(outputs.logits.shape       ,  lowercase__		)
		self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]       ,  lowercase__       ,  atol=1E-4		)		)

		# verify the loss (`config.norm_pix_loss` = `True`)
		__UpperCAmelCase							=      torch.tensor([0.5_142]       ,  device=lowercase__		)
		self.assertTrue(torch.allclose(outputs.loss       ,  lowercase__       ,  atol=1E-4		)		)

		# verify the loss (`config.norm_pix_loss` = `False`)
		__UpperCAmelCase							=      VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''       ,  norm_pix_loss=lowercase__		).to(
		    lowercase__		)

		with torch.no_grad():
			__UpperCAmelCase							=      model(**lowercase__		)

		__UpperCAmelCase							=      torch.tensor(torch.tensor([0.6_469]		)       ,  device=lowercase__		)
		self.assertTrue(torch.allclose(outputs.loss       ,  lowercase__       ,  atol=1E-4		)		)



'''simple docstring'''






import pytest
import requests

from datasets.utils.file_utils import http_head

from .utils import OfflineSimulationMode, RequestWouldHangIndefinitelyError, offline


@pytest.mark.integration
def 		_A     (    ):






 """simple docstring"""



 with offline(OfflineSimulationMode.CONNECTION_TIMES_OUT      ):
  with pytest.raises(A__      ):
   requests.request('''GET'''     ,				'''https://huggingface.co'''      )
  with pytest.raises(requests.exceptions.ConnectTimeout      ):
   requests.request('''GET'''     ,				'''https://huggingface.co'''     ,				timeout=1.0      )


@pytest.mark.integration
def 		_A     (    ):






 """simple docstring"""



 with offline(OfflineSimulationMode.CONNECTION_FAILS      ):
  with pytest.raises(requests.exceptions.ConnectionError      ):
   requests.request('''GET'''     ,				'''https://huggingface.co'''      )






def 		_A     (    ):






 """simple docstring"""



 with offline(OfflineSimulationMode.HF_DATASETS_OFFLINE_SET_TO_1      ):
  with pytest.raises(A__      ):
   http_head('''https://huggingface.co'''      )

