



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 				_a						:





							'''simple docstring'''

							def __init__(							self ,					__UpperCAmelCase ,					__UpperCAmelCase=13 ,					__UpperCAmelCase=10 ,					__UpperCAmelCase=3 ,					__UpperCAmelCase=2 ,					__UpperCAmelCase=2 ,					__UpperCAmelCase=2 ,					__UpperCAmelCase=True ,					__UpperCAmelCase=True ,					__UpperCAmelCase=32 ,					__UpperCAmelCase=5 ,					__UpperCAmelCase=4 ,					__UpperCAmelCase=37 ,					__UpperCAmelCase="gelu" ,					__UpperCAmelCase=0.1 ,					__UpperCAmelCase=0.1 ,					__UpperCAmelCase=10 ,					__UpperCAmelCase=0.0_2 ,					__UpperCAmelCase=0.9 ,					__UpperCAmelCase=None ,					):







											"""simple docstring"""



											a__	:							Union[str, Any]      						= parent
											a__	:							List[Any]      						= batch_size
											a__	:							Any      						= image_size
											a__	:							Any      						= num_channels
											a__	:							Optional[Any]      						= patch_size
											a__	:							Union[str, Any]      						= tubelet_size
											a__	:							Union[str, Any]      						= num_frames
											a__	:							List[Any]      						= is_training
											a__	:							List[str]      						= use_labels
											a__	:							Optional[int]      						= hidden_size
											a__	:							Tuple      						= num_hidden_layers
											a__	:							List[Any]      						= num_attention_heads
											a__	:							Any      						= intermediate_size
											a__	:							Optional[Any]      						= hidden_act
											a__	:							int      						= hidden_dropout_prob
											a__	:							Optional[int]      						= attention_probs_dropout_prob
											a__	:							Optional[Any]      						= type_sequence_label_size
											a__	:							Any      						= initializer_range
											a__	:							Any      						= mask_ratio
											a__	:							Optional[Any]      						= scope

											# in VideoMAE, the number of tokens equals num_frames/tubelet_size * num_patches per frame
											a__	:							Dict      						= (image_size // patch_size) ** 2
											a__	:							Optional[Any]      						= (num_frames // tubelet_size) * self.num_patches_per_frame

											# use this variable to define bool_masked_pos
											a__	:							List[Any]      						= int(mask_ratio * self.seq_length				)



							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							Any      						= floats_tensor(
											    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]				)

											a__	:							Optional[Any]      						= None
											if self.use_labels:
															a__	:							Union[str, Any]      						= ids_tensor([self.batch_size] ,					self.type_sequence_label_size				)

											a__	:							int      						= self.get_config()

											return config, pixel_values, labels



							def 		_A	(							self				):







											"""simple docstring"""



											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=__UpperCAmelCase ,					initializer_range=self.initializer_range ,					)



							def 		_A	(							self ,					__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase				):







											"""simple docstring"""



											a__	:							List[str]      						= VideoMAEModel(config=__UpperCAmelCase				)
											model.to(__UpperCAmelCase				)
											model.eval()
											a__	:							Union[str, Any]      						= model(__UpperCAmelCase				)
											self.parent.assertEqual(result.last_hidden_state.shape ,					(self.batch_size, self.seq_length, self.hidden_size)				)



							def 		_A	(							self ,					__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase				):







											"""simple docstring"""



											a__	:							Tuple      						= VideoMAEForPreTraining(__UpperCAmelCase				)
											model.to(__UpperCAmelCase				)
											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
											a__	:							Optional[Any]      						= torch.ones((self.num_masks,)				)
											a__	:							str      						= torch.cat([mask, torch.zeros(self.seq_length - mask.size(0				)				)]				)
											a__	:							str      						= mask.expand(self.batch_size ,					-1				).bool()

											a__	:							List[str]      						= model(__UpperCAmelCase ,					__UpperCAmelCase				)
											# model only returns predictions for masked patches
											a__	:							List[Any]      						= mask.sum().item()
											a__	:							int      						= 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				):







											"""simple docstring"""



											a__	:							str      						= self.prepare_config_and_inputs()
											a__						,   a__						,   a__	:							List[Any]      						= config_and_inputs
											a__	:							Dict      						= {"pixel_values": pixel_values}
											return config, inputs_dict





@require_torch
class 				_a						(       SCREAMING_SNAKE_CASE       ,  SCREAMING_SNAKE_CASE       ,  unittest.TestCase			):





							'''simple docstring'''

							A				:List[Any] =       (
							    (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
							)
							A				:Union[str, Any] =       (
							    {"feature-extraction": VideoMAEModel, "video-classification": VideoMAEForVideoClassification}
							    if is_torch_available()
							    else {}
							)

							A				:str =       False
							A				:Union[str, Any] =       False
							A				:Dict =       False
							A				:Any =       False



							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							Tuple      						= VideoMAEModelTester(self				)
											a__	:							Union[str, Any]      						= ConfigTester(self ,					config_class=__UpperCAmelCase ,					has_text_modality=__UpperCAmelCase ,					hidden_size=37				)



							def 		_A	(							self ,					__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase=False				):







											"""simple docstring"""



											a__	:							List[str]      						= copy.deepcopy(__UpperCAmelCase				)

											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
															a__	:							Tuple      						= torch.ones((self.model_tester.num_masks,)				)
															a__	:							List[str]      						= torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0				)				)]				)
															a__	:							Tuple      						= mask.expand(self.model_tester.batch_size ,					-1				).bool()
															a__	:							int      						= bool_masked_pos.to(__UpperCAmelCase				)

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

											return inputs_dict



							def 		_A	(							self				):







											"""simple docstring"""



											self.config_tester.run_common_tests()



							@unittest.skip(reason="VideoMAE does not use inputs_embeds"				)
							def 		_A	(							self				):







											"""simple docstring"""



											pass



							def 		_A	(							self				):







											"""simple docstring"""



											a__						,   a__	:							List[Any]      						= self.model_tester.prepare_config_and_inputs_for_common()

											for model_class in self.all_model_classes:
															a__	:							Dict      						= model_class(__UpperCAmelCase				)
															self.assertIsInstance(model.get_input_embeddings() ,					(nn.Module)				)
															a__	:							Optional[Any]      						= model.get_output_embeddings()
															self.assertTrue(x is None or isinstance(__UpperCAmelCase ,					nn.Linear				)				)



							def 		_A	(							self				):







											"""simple docstring"""



											a__						,   a__	:							Optional[int]      						= self.model_tester.prepare_config_and_inputs_for_common()

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

															a__	:							Dict      						= ["pixel_values"]
															self.assertListEqual(arg_names[:1] ,					__UpperCAmelCase				)



							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							Tuple      						= self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_model(*__UpperCAmelCase				)



							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							List[Any]      						= self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_for_pretraining(*__UpperCAmelCase				)



							@slow
							def 		_A	(							self				):







											"""simple docstring"""



											for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
															a__	:							Tuple      						= VideoMAEModel.from_pretrained(__UpperCAmelCase				)
															self.assertIsNotNone(__UpperCAmelCase				)



							def 		_A	(							self				):







											"""simple docstring"""



											if not self.has_attentions:
															pass

											else:
															a__						,   a__	:							Tuple      						= self.model_tester.prepare_config_and_inputs_for_common()
															a__	:							str      						= True

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

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

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

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

																			# Check attention is always last and order is fine
																			a__	:							int      						= True
																			a__	:							List[Any]      						= True
																			a__	:							Dict      						= model_class(__UpperCAmelCase				)
																			model.to(__UpperCAmelCase				)
																			model.eval()
																			with torch.no_grad():
																							a__	:							Optional[int]      						= model(**self._prepare_for_class(__UpperCAmelCase ,					__UpperCAmelCase				)				)

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

																			a__	:							List[str]      						= outputs.attentions

																			self.assertEqual(len(__UpperCAmelCase				) ,					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				):







											"""simple docstring"""



											def check_hidden_states_output(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase				):
															a__	:							Union[str, Any]      						= model_class(__UpperCAmelCase				)
															model.to(__UpperCAmelCase				)
															model.eval()

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

															a__	:							Tuple      						= outputs.hidden_states
															a__	:							Optional[Any]      						= self.model_tester.num_hidden_layers + 1
															self.assertEqual(len(__UpperCAmelCase				) ,					__UpperCAmelCase				)

															a__	:							str      						= self.model_tester.seq_length - self.model_tester.num_masks
															a__	:							Tuple      						= 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] ,					)

											a__						,   a__	:							List[str]      						= self.model_tester.prepare_config_and_inputs_for_common()

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

															# check that output_hidden_states also work using config
															del inputs_dict["output_hidden_states"]
															a__	:							Optional[Any]      						= True

															check_hidden_states_output(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase				)




							@unittest.skip("Will be fixed soon by reducing the size of the model used for common tests."				)
							def 		_A	(							self				):







											"""simple docstring"""



											pass






def     SCREAMING_SNAKE_CASE(      )	->						List[Any]:
				a__	:							str      						= hf_hub_download(
				    repo_id="hf-internal-testing/spaghetti-video"	,				filename="eating_spaghetti.npy"	,				repo_type="dataset" )
				a__	:							List[str]      						= np.load(__UpperCamelCase )
				return list(__UpperCamelCase )






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





							'''simple docstring'''

							@cached_property
							def 		_A	(							self				):







											"""simple docstring"""



											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				):







											"""simple docstring"""



											a__	:							List[str]      						= VideoMAEForVideoClassification.from_pretrained("MCG-NJU/videomae-base-finetuned-kinetics"				).to(
											    __UpperCAmelCase				)

											a__	:							Tuple      						= self.default_image_processor
											a__	:							Dict      						= prepare_video()
											a__	:							str      						= image_processor(__UpperCAmelCase ,					return_tensors="pt"				).to(__UpperCAmelCase				)

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

											# verify the logits
											a__	:							Tuple      						= torch.Size((1, 400)				)
											self.assertEqual(outputs.logits.shape ,					__UpperCAmelCase				)

											a__	:							Tuple      						= torch.tensor([0.3_6_6_9, -0.0_6_8_8, -0.2_4_2_1]				).to(__UpperCAmelCase				)

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




							@slow
							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							Optional[int]      						= VideoMAEForPreTraining.from_pretrained("MCG-NJU/videomae-base-short"				).to(__UpperCAmelCase				)

											a__	:							List[str]      						= self.default_image_processor
											a__	:							Tuple      						= prepare_video()
											a__	:							Dict      						= image_processor(__UpperCAmelCase ,					return_tensors="pt"				).to(__UpperCAmelCase				)

											# add boolean mask, indicating which patches to mask
											a__	:							List[Any]      						= hf_hub_download(repo_id="hf-internal-testing/bool-masked-pos" ,					filename="bool_masked_pos.pt"				)
											a__	:							Tuple      						= torch.load(__UpperCAmelCase				)

											# forward pass
											with torch.no_grad():
															a__	:							Optional[int]      						= model(**__UpperCAmelCase				)

											# verify the logits
											a__	:							Union[str, Any]      						= torch.Size([1, 1408, 1536]				)
											a__	:							Dict      						= torch.tensor(
											    [[0.7_9_9_4, 0.9_6_1_2, 0.8_5_0_8], [0.7_4_0_1, 0.8_9_5_8, 0.8_3_0_2], [0.5_8_6_2, 0.7_4_6_8, 0.7_3_2_5]] ,					device=__UpperCAmelCase				)
											self.assertEqual(outputs.logits.shape ,					__UpperCAmelCase				)
											self.assertTrue(torch.allclose(outputs.logits[0, :3, :3] ,					__UpperCAmelCase ,					atol=1E-4				)				)

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

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

											with torch.no_grad():
															a__	:							Tuple      						= model(**__UpperCAmelCase				)

											a__	:							Optional[int]      						= torch.tensor(torch.tensor([0.6_4_6_9]				) ,					device=__UpperCAmelCase				)
											self.assertTrue(torch.allclose(outputs.loss ,					__UpperCAmelCase ,					atol=1E-4				)				)







from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


lowerCamelCase		       =      {
    """configuration_instructblip""": [
        """INSTRUCTBLIP_PRETRAINED_CONFIG_ARCHIVE_MAP""",
        """InstructBlipConfig""",
        """InstructBlipQFormerConfig""",
        """InstructBlipVisionConfig""",
    ],
    """processing_instructblip""": ["""InstructBlipProcessor"""],
}

try:
						if not is_torch_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						lowerCamelCase		       =      [
						    """INSTRUCTBLIP_PRETRAINED_MODEL_ARCHIVE_LIST""",
						    """InstructBlipQFormerModel""",
						    """InstructBlipPreTrainedModel""",
						    """InstructBlipForConditionalGeneration""",
						    """InstructBlipVisionModel""",
						]

if TYPE_CHECKING:
						from .configuration_instructblip import (
						    INSTRUCTBLIP_PRETRAINED_CONFIG_ARCHIVE_MAP,
						    InstructBlipConfig,
						    InstructBlipQFormerConfig,
						    InstructBlipVisionConfig,
						)
						from .processing_instructblip import InstructBlipProcessor

						try:
												if not is_torch_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_instructblip import (
												    INSTRUCTBLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
												    InstructBlipForConditionalGeneration,
												    InstructBlipPreTrainedModel,
												    InstructBlipQFormerModel,
												    InstructBlipVisionModel,
												)

else:
						import sys

						lowerCamelCase		       =      _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)


