


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     :

				"""simple docstring"""







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

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

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

				def   _UpperCAmelCase       (    self )						->					Tuple:
					A             =					floats_tensor(
					    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size] )

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

					A             =					self.get_config()

					return config, pixel_values, labels

				def   _UpperCAmelCase       (    self )						->					Any:
					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=a__					, initializer_range=self.initializer_range					, )

				def   _UpperCAmelCase       (    self					, a__					, a__					, a__ )						->					List[str]:
					A             =					VideoMAEModel(config=a__ )
					model.to(a__ )
					model.eval()
					A             =					model(a__ )
					self.parent.assertEqual(result.last_hidden_state.shape					, (self.batch_size, self.seq_length, self.hidden_size) )

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

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

				def   _UpperCAmelCase       (    self )						->					Dict:
					A             =					self.prepare_config_and_inputs()
					A       ,      A       ,      A             =					config_and_inputs
					A             =					{"""pixel_values""": pixel_values}
					return config, inputs_dict






@require_torch
class    _UpperCamelCase     (       __snake_case     ,		__snake_case     ,		unittest.TestCase					):

				"""simple docstring"""







				lowerCAmelCase					=			(
				    (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
				)
				lowerCAmelCase					=			(
				    {'feature-extraction': VideoMAEModel, 'video-classification': VideoMAEForVideoClassification}
				    if is_torch_available()
				    else {}
				)

				lowerCAmelCase					=			False
				lowerCAmelCase					=			False
				lowerCAmelCase					=			False
				lowerCAmelCase					=			False

				def   _UpperCAmelCase       (    self )						->					Any:
					A             =					VideoMAEModelTester(self )
					A             =					ConfigTester(self					, config_class=a__					, has_text_modality=a__					, hidden_size=37 )

				def   _UpperCAmelCase       (    self					, a__					, a__					, a__=False )						->					str:
					A             =					copy.deepcopy(a__ )

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

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

					return inputs_dict

				def   _UpperCAmelCase       (    self )						->					Tuple:
					self.config_tester.run_common_tests()

				@unittest.skip(reason="""VideoMAE does not use inputs_embeds""" )
				def   _UpperCAmelCase       (    self )						->					List[str]:
					pass

				def   _UpperCAmelCase       (    self )						->					List[str]:
					A       ,      A             =					self.model_tester.prepare_config_and_inputs_for_common()

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

				def   _UpperCAmelCase       (    self )						->					Optional[Any]:
					A       ,      A             =					self.model_tester.prepare_config_and_inputs_for_common()

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

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

				def   _UpperCAmelCase       (    self )						->					int:
					A             =					self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_model(*a__ )

				def   _UpperCAmelCase       (    self )						->					List[str]:
					A             =					self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_for_pretraining(*a__ )

				@slow
				def   _UpperCAmelCase       (    self )						->					str:
					for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
						A             =					VideoMAEModel.from_pretrained(a__ )
						self.assertIsNotNone(a__ )

				def   _UpperCAmelCase       (    self )						->					Union[str, Any]:
					if not self.has_attentions:
						pass

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

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

							A             =					True
							A             =					False
							A             =					True
							A             =					model_class(a__ )
							model.to(a__ )
							model.eval()
							with torch.no_grad():
								A             =					model(**self._prepare_for_class(a__					, a__ ) )
							A             =					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"]
							A             =					True
							A             =					model_class(a__ )
							model.to(a__ )
							model.eval()
							with torch.no_grad():
								A             =					model(**self._prepare_for_class(a__					, a__ ) )
							A             =					outputs.attentions
							self.assertEqual(len(a__ )					, self.model_tester.num_hidden_layers )

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

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

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

							A             =					outputs.attentions

							self.assertEqual(len(a__ )					, 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   _UpperCAmelCase       (    self )						->					Optional[int]:
					def check_hidden_states_output(a__					, a__					, a__ ):
						A             =					model_class(a__ )
						model.to(a__ )
						model.eval()

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

						A             =					outputs.hidden_states
						A             =					self.model_tester.num_hidden_layers + 1
						self.assertEqual(len(a__ )					, a__ )

						A             =					self.model_tester.seq_length - self.model_tester.num_masks
						A             =					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             =					self.model_tester.prepare_config_and_inputs_for_common()

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

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

						check_hidden_states_output(a__					, a__					, a__ )

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





def 	_lowerCAmelCase					( )			-> Any:





	"""simple docstring"""







	A             =					hf_hub_download(
	    repo_id="""hf-internal-testing/spaghetti-video"""				,       filename="""eating_spaghetti.npy"""				,       repo_type="""dataset"""    )
	A             =					np.load(lowerCAmelCase__    )
	return list(lowerCAmelCase__    )






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

				"""simple docstring"""







				@cached_property
				def   _UpperCAmelCase       (    self )						->					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   _UpperCAmelCase       (    self )						->					Tuple:
					A             =					VideoMAEForVideoClassification.from_pretrained("""MCG-NJU/videomae-base-finetuned-kinetics""" ).to(
					    a__ )

					A             =					self.default_image_processor
					A             =					prepare_video()
					A             =					image_processor(a__					, return_tensors="""pt""" ).to(a__ )

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

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

					A             =					torch.tensor([0.36_69, -0.06_88, -0.24_21] ).to(a__ )

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

				@slow
				def   _UpperCAmelCase       (    self )						->					Union[str, Any]:
					A             =					VideoMAEForPreTraining.from_pretrained("""MCG-NJU/videomae-base-short""" ).to(a__ )

					A             =					self.default_image_processor
					A             =					prepare_video()
					A             =					image_processor(a__					, return_tensors="""pt""" ).to(a__ )

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

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

					# verify the logits
					A             =					torch.Size([1, 1408, 1536] )
					A             =					torch.tensor(
					    [[0.79_94, 0.96_12, 0.85_08], [0.74_01, 0.89_58, 0.83_02], [0.58_62, 0.74_68, 0.73_25]]					, device=a__ )
					self.assertEqual(outputs.logits.shape					, a__ )
					self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]					, a__					, atol=1e-4 ) )

					# verify the loss (`config.norm_pix_loss` = `True`)
					A             =					torch.tensor([0.51_42]					, device=a__ )
					self.assertTrue(torch.allclose(outputs.loss					, a__					, atol=1e-4 ) )

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

					with torch.no_grad():
						A             =					model(**a__ )

					A             =					torch.tensor(torch.tensor([0.64_69] )					, device=a__ )
					self.assertTrue(torch.allclose(outputs.loss					, a__					, atol=1e-4 ) )





"""simple docstring"""


import os
from collections import namedtuple

import pytest

from datasets import ClassLabel, Features, Sequence, Value
from datasets.commands.test import TestCommand
from datasets.info import DatasetInfo, DatasetInfosDict


A_    							=					namedtuple(
    """_TestCommandArgs""",
    [
        """dataset""",
        """name""",
        """cache_dir""",
        """data_dir""",
        """all_configs""",
        """save_infos""",
        """ignore_verifications""",
        """force_redownload""",
        """clear_cache""",
    ],
    defaults=[None, None, None, False, False, False, False, False],
)






def  lowercase						(	lowerCAmelCase__	,lowerCAmelCase__						):
			return (abs(source - target						) / target) < 0.01






@pytest.mark.integration
def  lowercase						(	lowerCAmelCase__						):
			lowerCamelCase_					=  _TestCommandArgs(dataset=lowerCAmelCase__	,all_configs=lowerCAmelCase__	,save_infos=lowerCAmelCase__						)
			lowerCamelCase_					=  TestCommand(*lowerCAmelCase__						)
			test_command.run()
			lowerCamelCase_					=  os.path.join(lowerCAmelCase__	,'''README.md'''						)
			assert os.path.exists(lowerCAmelCase__						)
			lowerCamelCase_					=  DatasetInfosDict.from_directory(lowerCAmelCase__						)
			lowerCamelCase_					=  DatasetInfosDict(
			    {
			        '''default''': DatasetInfo(
			            features=Features(
			                {
			                    '''tokens''': Sequence(Value('''string'''						)						),
			                    '''ner_tags''': Sequence(
			                        ClassLabel(names=['''O''', '''B-PER''', '''I-PER''', '''B-ORG''', '''I-ORG''', '''B-LOC''', '''I-LOC''']						)						),
			                    '''langs''': Sequence(Value('''string'''						)						),
			                    '''spans''': Sequence(Value('''string'''						)						),
			                }						)	,splits=[
			                {
			                    '''name''': '''train''',
			                    '''num_bytes''': 2_351_563,
			                    '''num_examples''': 10_000,
			                },
			                {
			                    '''name''': '''validation''',
			                    '''num_bytes''': 238_418,
			                    '''num_examples''': 1_000,
			                },
			            ]	,download_size=3_940_680	,dataset_size=2_589_981	,)
			    }						)
			assert dataset_infos.keys() == expected_dataset_infos.keys()
			for key in DatasetInfo._INCLUDED_INFO_IN_YAML:
						lowerCamelCase_		, lowerCamelCase_					=  getattr(dataset_infos['''default''']	,lowerCAmelCase__						), getattr(expected_dataset_infos['''default''']	,lowerCAmelCase__						)
						if key == "num_bytes":
									assert is_apercent_close(lowerCAmelCase__	,lowerCAmelCase__						)
						elif key == "splits":
									assert list(lowerCAmelCase__						) == list(lowerCAmelCase__						)
									for split in result:
												assert result[split].name == expected[split].name
												assert result[split].num_examples == expected[split].num_examples
												assert is_apercent_close(result[split].num_bytes	,expected[split].num_bytes						)
						else:
									result == expected



