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



							'''simple docstring'''







							def __init__(     self  ,    _lowerCamelCase  ,    _lowerCamelCase=13  ,    _lowerCamelCase=10  ,    _lowerCamelCase=3  ,    _lowerCamelCase=2  ,    _lowerCamelCase=2  ,    _lowerCamelCase=2  ,    _lowerCamelCase=True  ,    _lowerCamelCase=True  ,    _lowerCamelCase=32  ,    _lowerCamelCase=5  ,    _lowerCamelCase=4  ,    _lowerCamelCase=37  ,    _lowerCamelCase="gelu"  ,    _lowerCamelCase=0.1  ,    _lowerCamelCase=0.1  ,    _lowerCamelCase=10  ,    _lowerCamelCase=0.02  ,    _lowerCamelCase=0.9  ,    _lowerCamelCase=None  ,    )			->							List[Any]:




										'''simple docstring'''
										__lowercase       =       parent
										__lowercase       =       batch_size
										__lowercase       =       image_size
										__lowercase       =       num_channels
										__lowercase       =       patch_size
										__lowercase       =       tubelet_size
										__lowercase       =       num_frames
										__lowercase       =       is_training
										__lowercase       =       use_labels
										__lowercase       =       hidden_size
										__lowercase       =       num_hidden_layers
										__lowercase       =       num_attention_heads
										__lowercase       =       intermediate_size
										__lowercase       =       hidden_act
										__lowercase       =       hidden_dropout_prob
										__lowercase       =       attention_probs_dropout_prob
										__lowercase       =       type_sequence_label_size
										__lowercase       =       initializer_range
										__lowercase       =       mask_ratio
										__lowercase       =       scope

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

										# use this variable to define bool_masked_pos
										__lowercase       =       int(mask_ratio * self.seq_length       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Union[str, Any]:




										'''simple docstring'''
										__lowercase       =       floats_tensor(
										    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]       )

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

										__lowercase       =       self.get_config()

										return config, pixel_values, labels
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Optional[Any]:




										'''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=_lowerCamelCase  ,    initializer_range=self.initializer_range  ,    )
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase       )			->							Any:




										'''simple docstring'''
										__lowercase       =       VideoMAEModel(config=_lowerCamelCase       )
										model.to(_lowerCamelCase       )
										model.eval()
										__lowercase       =       model(_lowerCamelCase       )
										self.parent.assertEqual(result.last_hidden_state.shape  ,    (self.batch_size, self.seq_length, self.hidden_size)       )
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase       )			->							Any:




										'''simple docstring'''
										__lowercase       =       VideoMAEForPreTraining(_lowerCamelCase       )
										model.to(_lowerCamelCase       )
										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
										__lowercase       =       torch.ones((self.num_masks,)       )
										__lowercase       =       torch.cat([mask, torch.zeros(self.seq_length - mask.size(0       )       )]       )
										__lowercase       =       mask.expand(self.batch_size  ,    -1       ).bool()

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





							def   SCREAMING_SNAKE_CASE						(     self       )			->							Dict:




										'''simple docstring'''
										__lowercase       =       self.prepare_config_and_inputs()
										__lowercase   ,					__lowercase   ,					__lowercase       =       config_and_inputs
										__lowercase       =       {"pixel_values": pixel_values}
										return config, inputs_dict





@require_torch
class 					__a     ( __a					,   __a					,   unittest.TestCase       ):



							'''simple docstring'''







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

							_lowerCamelCase					:			Optional[Any]          =   False
							_lowerCamelCase					:			List[str]          =   False
							_lowerCamelCase					:			Union[str, Any]          =   False
							_lowerCamelCase					:			Dict          =   False
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Any:




										'''simple docstring'''
										__lowercase       =       VideoMAEModelTester(self       )
										__lowercase       =       ConfigTester(self  ,    config_class=_lowerCamelCase  ,    has_text_modality=_lowerCamelCase  ,    hidden_size=37       )
							def   SCREAMING_SNAKE_CASE						(     self  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase=False       )			->							str:




										'''simple docstring'''
										__lowercase       =       copy.deepcopy(_lowerCamelCase       )

										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
													__lowercase       =       torch.ones((self.model_tester.num_masks,)       )
													__lowercase       =       torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0       )       )]       )
													__lowercase       =       mask.expand(self.model_tester.batch_size  ,    -1       ).bool()
													__lowercase       =       bool_masked_pos.to(_lowerCamelCase       )

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

										return inputs_dict
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Any:




										'''simple docstring'''
										self.config_tester.run_common_tests()
							@unittest.skip(reason="VideoMAE does not use inputs_embeds"       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							List[Any]:




										'''simple docstring'''
										pass
							def   SCREAMING_SNAKE_CASE						(     self       )			->							List[Any]:




										'''simple docstring'''
										__lowercase   ,					__lowercase       =       self.model_tester.prepare_config_and_inputs_for_common()

										for model_class in self.all_model_classes:
													__lowercase       =       model_class(_lowerCamelCase       )
													self.assertIsInstance(model.get_input_embeddings()  ,    (nn.Module)       )
													__lowercase       =       model.get_output_embeddings()
													self.assertTrue(x is None or isinstance(_lowerCamelCase  ,    nn.Linear       )       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							List[Any]:




										'''simple docstring'''
										__lowercase   ,					__lowercase       =       self.model_tester.prepare_config_and_inputs_for_common()

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

													__lowercase       =       ["pixel_values"]
													self.assertListEqual(arg_names[:1]  ,    _lowerCamelCase       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Any:




										'''simple docstring'''
										__lowercase       =       self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_model(*_lowerCamelCase       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							List[Any]:




										'''simple docstring'''
										__lowercase       =       self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_for_pretraining(*_lowerCamelCase       )
							@slow
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Dict:




										'''simple docstring'''
										for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
													__lowercase       =       VideoMAEModel.from_pretrained(_lowerCamelCase       )
													self.assertIsNotNone(_lowerCamelCase       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Tuple:




										'''simple docstring'''
										if not self.has_attentions:
													pass

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

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

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

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

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

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

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

																__lowercase       =       outputs.attentions

																self.assertEqual(len(_lowerCamelCase       )  ,    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   SCREAMING_SNAKE_CASE						(     self       )			->							List[str]:




										'''simple docstring'''
										def check_hidden_states_output(_lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase       ):
													__lowercase       =       model_class(_lowerCamelCase       )
													model.to(_lowerCamelCase       )
													model.eval()

													with torch.no_grad():
																__lowercase       =       model(**self._prepare_for_class(_lowerCamelCase  ,    _lowerCamelCase       )       )

													__lowercase       =       outputs.hidden_states
													__lowercase       =       self.model_tester.num_hidden_layers + 1
													self.assertEqual(len(_lowerCamelCase       )  ,    _lowerCamelCase       )

													__lowercase       =       self.model_tester.seq_length - self.model_tester.num_masks
													__lowercase       =       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]  ,    )

										__lowercase   ,					__lowercase       =       self.model_tester.prepare_config_and_inputs_for_common()

										for model_class in self.all_model_classes:
													__lowercase       =       True
													check_hidden_states_output(_lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase       )

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

													check_hidden_states_output(_lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase       )





							@unittest.skip("Will be fixed soon by reducing the size of the model used for common tests."       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							List[Any]:




										'''simple docstring'''
										pass





def      lowerCAmelCase__			(		)						->Optional[int]:
			__lowercase       =       hf_hub_download(
			    repo_id="hf-internal-testing/spaghetti-video"			,   filename="eating_spaghetti.npy"			,   repo_type="dataset"      )
			__lowercase       =       np.load(__magic_name__      )
			return list(__magic_name__      )







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



							'''simple docstring'''







							@cached_property
							def   SCREAMING_SNAKE_CASE						(     self       )			->							str:




										'''simple docstring'''
										# 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   SCREAMING_SNAKE_CASE						(     self       )			->							List[str]:




										'''simple docstring'''
										__lowercase       =       VideoMAEForVideoClassification.from_pretrained("MCG-NJU/videomae-base-finetuned-kinetics"       ).to(
										    _lowerCamelCase       )

										__lowercase       =       self.default_image_processor
										__lowercase       =       prepare_video()
										__lowercase       =       image_processor(_lowerCamelCase  ,    return_tensors="pt"       ).to(_lowerCamelCase       )

										# forward pass
										with torch.no_grad():
													__lowercase       =       model(**_lowerCamelCase       )

										# verify the logits
										__lowercase       =       torch.Size((1, 400)       )
										self.assertEqual(outputs.logits.shape  ,    _lowerCamelCase       )

										__lowercase       =       torch.tensor([0.3669, -0.0688, -0.2421]       ).to(_lowerCamelCase       )

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





							@slow
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Tuple:




										'''simple docstring'''
										__lowercase       =       VideoMAEForPreTraining.from_pretrained("MCG-NJU/videomae-base-short"       ).to(_lowerCamelCase       )

										__lowercase       =       self.default_image_processor
										__lowercase       =       prepare_video()
										__lowercase       =       image_processor(_lowerCamelCase  ,    return_tensors="pt"       ).to(_lowerCamelCase       )

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

										# forward pass
										with torch.no_grad():
													__lowercase       =       model(**_lowerCamelCase       )

										# verify the logits
										__lowercase       =       torch.Size([1, 1_408, 1_536]       )
										__lowercase       =       torch.tensor(
										    [[0.7994, 0.9612, 0.8508], [0.7401, 0.8958, 0.8302], [0.5862, 0.7468, 0.7325]]  ,    device=_lowerCamelCase       )
										self.assertEqual(outputs.logits.shape  ,    _lowerCamelCase       )
										self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]  ,    _lowerCamelCase  ,    atol=1e-4       )       )

										# verify the loss (`config.norm_pix_loss` = `True`)
										__lowercase       =       torch.tensor([0.5142]  ,    device=_lowerCamelCase       )
										self.assertTrue(torch.allclose(outputs.loss  ,    _lowerCamelCase  ,    atol=1e-4       )       )

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

										with torch.no_grad():
													__lowercase       =       model(**_lowerCamelCase       )

										__lowercase       =       torch.tensor(torch.tensor([0.6469]       )  ,    device=_lowerCamelCase       )
										self.assertTrue(torch.allclose(outputs.loss  ,    _lowerCamelCase  ,    atol=1e-4       )       )







"""simple docstring"""






import enum
import os
from hashlib import shaaaa
from typing import Optional

from .. import config
from .logging import get_logger


_lowercase							=				get_logger(__name__)

class 					__a     ( enum.Enum       ):



							'''simple docstring'''







							_lowerCamelCase					:			Tuple          =   """all_checks"""
							_lowerCamelCase					:			Optional[int]          =   """basic_checks"""
							_lowerCamelCase					:			str          =   """no_checks"""





class 					__a     ( __a       ):



							'''simple docstring'''












class 					__a     ( __a       ):



							'''simple docstring'''












class 					__a     ( __a       ):



							'''simple docstring'''












class 					__a     ( __a       ):



							'''simple docstring'''












def      lowerCAmelCase__			(		__magic_name__			,   __magic_name__			,   __magic_name__=None      )						->Union[str, Any]:
			if expected_checksums is None:
						logger.info("Unable to verify checksums."      )
						return
			if len(set(__magic_name__      ) - set(__magic_name__      )      ) > 0:
						raise ExpectedMoreDownloadedFiles(str(set(__magic_name__      ) - set(__magic_name__      )      )      )
			if len(set(__magic_name__      ) - set(__magic_name__      )      ) > 0:
						raise UnexpectedDownloadedFile(str(set(__magic_name__      ) - set(__magic_name__      )      )      )
			__lowercase       =       [url for url in expected_checksums if expected_checksums[url] != recorded_checksums[url]]
			__lowercase       =       " for " + verification_name if verification_name is not None else ""
			if len(__magic_name__      ) > 0:
						raise NonMatchingChecksumError(
						    F'''Checksums didn\'t match{for_verification_name}:\n'''
						    F'''{bad_urls}\n'''
						    "Set `verification_mode='no_checks'` to skip checksums verification and ignore this error"      )
			logger.info("All the checksums matched successfully" + for_verification_name      )

class 					__a     ( __a       ):



							'''simple docstring'''












class 					__a     ( __a       ):



							'''simple docstring'''












class 					__a     ( __a       ):



							'''simple docstring'''












class 					__a     ( __a       ):



							'''simple docstring'''












def      lowerCAmelCase__			(		__magic_name__			,   __magic_name__      )						->Any:
			if expected_splits is None:
						logger.info("Unable to verify splits sizes."      )
						return
			if len(set(__magic_name__      ) - set(__magic_name__      )      ) > 0:
						raise ExpectedMoreSplits(str(set(__magic_name__      ) - set(__magic_name__      )      )      )
			if len(set(__magic_name__      ) - set(__magic_name__      )      ) > 0:
						raise UnexpectedSplits(str(set(__magic_name__      ) - set(__magic_name__      )      )      )
			__lowercase       =       [
			    {"expected": expected_splits[name], "recorded": recorded_splits[name]}
			    for name in expected_splits
			    if expected_splits[name].num_examples != recorded_splits[name].num_examples
			]
			if len(__magic_name__      ) > 0:
						raise NonMatchingSplitsSizesError(str(__magic_name__      )      )
			logger.info("All the splits matched successfully."      )



def      lowerCAmelCase__			(		__magic_name__			,   __magic_name__ = True      )						->dict:
			if record_checksum:
						__lowercase       =       shaaaa()
						with open(__magic_name__			,   "rb"      ) as f:
									for chunk in iter(lambda: f.read(1 << 2_0      )			,   b""      ):
												m.update(__magic_name__      )
									__lowercase       =       m.hexdigest()
			else:
						__lowercase       =       None
			return {"num_bytes": os.path.getsize(__magic_name__      ), "checksum": checksum}







def      lowerCAmelCase__			(		__magic_name__      )						->List[str]:
			if dataset_size and config.IN_MEMORY_MAX_SIZE:
						return dataset_size < config.IN_MEMORY_MAX_SIZE
			else:
						return False





