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




							def __init__(				self   ,A__   ,A__=1_3   ,A__=1_0   ,A__=3   ,A__=2   ,A__=2   ,A__=2   ,A__=True   ,A__=True   ,A__=3_2   ,A__=5   ,A__=4   ,A__=3_7   ,A__="gelu"   ,A__=0.1   ,A__=0.1   ,A__=1_0   ,A__=0.02   ,A__=0.9   ,A__=None   ,):
												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  A__      (				self):
												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  A__      (				self):
												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  A__      (				self   ,A__   ,A__   ,A__):
												lowercase							=						VideoMAEModel(config=A__)
												model.to(A__)
												model.eval()
												lowercase							=						model(A__)
												self.parent.assertEqual(result.last_hidden_state.shape   ,(self.batch_size, self.seq_length, self.hidden_size))




							def  A__      (				self   ,A__   ,A__   ,A__):
												lowercase							=						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
												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(A__   ,A__)
												# 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  A__      (				self):
												lowercase							=						self.prepare_config_and_inputs()
												lowercase	,						lowercase	,						lowercase							=						config_and_inputs
												lowercase							=						{'''pixel_values''': pixel_values}
												return config, inputs_dict







@require_torch
class 				lowercase		(					SCREAMING_SNAKE_CASE__       ,  SCREAMING_SNAKE_CASE__       ,  unittest.TestCase		):
							lowercase_       :       str       					=(
							    (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
							)
							lowercase_       :       Tuple       					=(
							    {'''feature-extraction''': VideoMAEModel, '''video-classification''': VideoMAEForVideoClassification}
							    if is_torch_available()
							    else {}
							)

							lowercase_       :       List[str]       					=False
							lowercase_       :       str       					=False
							lowercase_       :       int       					=False
							lowercase_       :       Dict       					=False




							def  A__      (				self):
												lowercase							=						VideoMAEModelTester(self)
												lowercase							=						ConfigTester(self   ,config_class=A__   ,has_text_modality=A__   ,hidden_size=3_7)




							def  A__      (				self   ,A__   ,A__   ,A__=False):
												lowercase							=						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
																	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(A__)

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

												return inputs_dict




							def  A__      (				self):
												self.config_tester.run_common_tests()




							@unittest.skip(reason='''VideoMAE does not use inputs_embeds''')
							def  A__      (				self):
												pass




							def  A__      (				self):
												lowercase	,						lowercase							=						self.model_tester.prepare_config_and_inputs_for_common()

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




							def  A__      (				self):
												lowercase	,						lowercase							=						self.model_tester.prepare_config_and_inputs_for_common()

												for model_class in self.all_model_classes:
																	lowercase							=						model_class(A__)
																	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]   ,A__)




							def  A__      (				self):
												lowercase							=						self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_model(*A__)




							def  A__      (				self):
												lowercase							=						self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_for_pretraining(*A__)




							@slow
							def  A__      (				self):
												for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																	lowercase							=						VideoMAEModel.from_pretrained(A__)
																	self.assertIsNotNone(A__)




							def  A__      (				self):
												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(A__)
																						model.to(A__)
																						model.eval()
																						with torch.no_grad():
																											lowercase							=						model(**self._prepare_for_class(A__   ,A__))
																						lowercase							=						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							=						True
																						lowercase							=						model_class(A__)
																						model.to(A__)
																						model.eval()
																						with torch.no_grad():
																											lowercase							=						model(**self._prepare_for_class(A__   ,A__))
																						lowercase							=						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]   ,)
																						lowercase							=						len(A__)

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

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

																						lowercase							=						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  A__      (				self):
												def check_hidden_states_output(A__   ,A__   ,A__):
																	lowercase							=						model_class(A__)
																	model.to(A__)
																	model.eval()

																	with torch.no_grad():
																						lowercase							=						model(**self._prepare_for_class(A__   ,A__))

																	lowercase							=						outputs.hidden_states
																	lowercase							=						self.model_tester.num_hidden_layers + 1
																	self.assertEqual(len(A__)   ,A__)

																	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(A__   ,A__   ,A__)

																	# check that output_hidden_states also work using config
																	del inputs_dict["output_hidden_states"]
																	lowercase							=						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  A__      (				self):
												pass



def        UpperCamelCase    (							):
					'''simple docstring'''
					lowercase							=						hf_hub_download(
					    repo_id='''hf-internal-testing/spaghetti-video''' ,			filename='''eating_spaghetti.npy''' ,			repo_type='''dataset'''		)
					lowercase							=						np.load(lowerCAmelCase__		)
					return list(lowerCAmelCase__		)







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




							@cached_property
							def  A__      (				self):
												# 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):
												lowercase							=						VideoMAEForVideoClassification.from_pretrained('''MCG-NJU/videomae-base-finetuned-kinetics''').to(
												    A__)

												lowercase							=						self.default_image_processor
												lowercase							=						prepare_video()
												lowercase							=						image_processor(A__   ,return_tensors='''pt''').to(A__)

												# forward pass
												with torch.no_grad():
																	lowercase							=						model(**A__)

												# verify the logits
												lowercase							=						torch.Size((1, 4_0_0))
												self.assertEqual(outputs.logits.shape   ,A__)

												lowercase							=						torch.tensor([0.3669, -0.0688, -0.2421]).to(A__)

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





							@slow
							def  A__      (				self):
												lowercase							=						VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short''').to(A__)

												lowercase							=						self.default_image_processor
												lowercase							=						prepare_video()
												lowercase							=						image_processor(A__   ,return_tensors='''pt''').to(A__)

												# 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(A__)

												# forward pass
												with torch.no_grad():
																	lowercase							=						model(**A__)

												# verify the logits
												lowercase							=						torch.Size([1, 1_4_0_8, 1_5_3_6])
												lowercase							=						torch.tensor(
												    [[0.7994, 0.9612, 0.8508], [0.7401, 0.8958, 0.8302], [0.5862, 0.7468, 0.7325]]   ,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`)
												lowercase							=						torch.tensor([0.5142]   ,device=A__)
												self.assertTrue(torch.allclose(outputs.loss   ,A__   ,atol=1E-4))

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

												with torch.no_grad():
																	lowercase							=						model(**A__)

												lowercase							=						torch.tensor(torch.tensor([0.6469])   ,device=A__)
												self.assertTrue(torch.allclose(outputs.loss   ,A__   ,atol=1E-4))






import argparse
import os

import numpy as np
import tensorflow as tf
import torch

from transformers import BertModel



def        UpperCamelCase    (							lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__		):
					'''simple docstring'''
					lowercase							=						('''dense.weight''', '''attention.self.query''', '''attention.self.key''', '''attention.self.value''')

					lowercase							=						(
					    ('''layer.''', '''layer_'''),
					    ('''word_embeddings.weight''', '''word_embeddings'''),
					    ('''position_embeddings.weight''', '''position_embeddings'''),
					    ('''token_type_embeddings.weight''', '''token_type_embeddings'''),
					    ('''.''', '''/'''),
					    ('''LayerNorm/weight''', '''LayerNorm/gamma'''),
					    ('''LayerNorm/bias''', '''LayerNorm/beta'''),
					    ('''weight''', '''kernel'''),
					)

					if not os.path.isdir(lowerCAmelCase__		):
										os.makedirs(lowerCAmelCase__		)

					lowercase							=						model.state_dict()

					def to_tf_var_name(lowerCAmelCase__		):
										for patt, repl in iter(lowerCAmelCase__		):
															lowercase							=						name.replace(lowerCAmelCase__ ,			lowerCAmelCase__		)
										return f'bert/{name}'

					def create_tf_var(lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__		):
										lowercase							=						tf.dtypes.as_dtype(tensor.dtype		)
										lowercase							=						tf.get_variable(dtype=lowerCAmelCase__ ,			shape=tensor.shape ,			name=lowerCAmelCase__ ,			initializer=tf.zeros_initializer()		)
										session.run(tf.variables_initializer([tf_var]		)		)
										session.run(lowerCAmelCase__		)
										return tf_var

					tf.reset_default_graph()
					with tf.Session() as session:
										for var_name in state_dict:
															lowercase							=						to_tf_var_name(lowerCAmelCase__		)
															lowercase							=						state_dict[var_name].numpy()
															if any(x in var_name for x in tensors_to_transpose		):
																				lowercase							=						torch_tensor.T
															lowercase							=						create_tf_var(tensor=lowerCAmelCase__ ,			name=lowerCAmelCase__ ,			session=lowerCAmelCase__		)
															tf.keras.backend.set_value(lowerCAmelCase__ ,			lowerCAmelCase__		)
															lowercase							=						session.run(lowerCAmelCase__		)
															print(f'Successfully created {tf_name}: {np.allclose(lowerCAmelCase__ ,			lowerCAmelCase__		)}'		)

										lowercase							=						tf.train.Saver(tf.trainable_variables()		)
										saver.save(lowerCAmelCase__ ,			os.path.join(lowerCAmelCase__ ,			model_name.replace('''-''' ,			'''_'''		) + '''.ckpt'''		)		)



def        UpperCamelCase    (							lowerCAmelCase__=None		):
					'''simple docstring'''
					lowercase							=						argparse.ArgumentParser()
					parser.add_argument('''--model_name''' ,			type=lowerCAmelCase__ ,			required=lowerCAmelCase__ ,			help='''model name e.g. bert-base-uncased'''		)
					parser.add_argument(
					    '''--cache_dir''' ,			type=lowerCAmelCase__ ,			default=lowerCAmelCase__ ,			required=lowerCAmelCase__ ,			help='''Directory containing pytorch model'''		)
					parser.add_argument('''--pytorch_model_path''' ,			type=lowerCAmelCase__ ,			required=lowerCAmelCase__ ,			help='''/path/to/<pytorch-model-name>.bin'''		)
					parser.add_argument('''--tf_cache_dir''' ,			type=lowerCAmelCase__ ,			required=lowerCAmelCase__ ,			help='''Directory in which to save tensorflow model'''		)
					lowercase							=						parser.parse_args(lowerCAmelCase__		)

					lowercase							=						BertModel.from_pretrained(
					    pretrained_model_name_or_path=args.model_name ,			state_dict=torch.load(args.pytorch_model_path		) ,			cache_dir=args.cache_dir ,			)

					convert_pytorch_checkpoint_to_tf(model=lowerCAmelCase__ ,			ckpt_dir=args.tf_cache_dir ,			model_name=args.model_name		)


if __name__ == "__main__":
							main()




