



"""simple docstring"""
import argparse
import json

import gdown
import numpy as np
import torch
from huggingface_hub import hf_hub_download

from transformers import TimesformerConfig, TimesformerForVideoClassification, VideoMAEImageProcessor



def 				lowerCAmelCase_(      lowercase_   :							Optional[int]							)		->				int:
				_lowerCamelCase					     = TimesformerConfig()

				if "large" in model_name:
								_lowerCamelCase					     = 96

				if "hr" in model_name:
								_lowerCamelCase					     = 16
								_lowerCamelCase					     = 4_48

				_lowerCamelCase					     = '''huggingface/label-files'''
				if "k400" in model_name:
								_lowerCamelCase					     = 4_00
								_lowerCamelCase					     = '''kinetics400-id2label.json'''
				elif "k600" in model_name:
								_lowerCamelCase					     = 6_00
								_lowerCamelCase					     = '''kinetics600-id2label.json'''
				elif "ssv2" in model_name:
								_lowerCamelCase					     = 1_74
								_lowerCamelCase					     = '''something-something-v2-id2label.json'''
				else:
								raise ValueError('''Model name should either contain \'k400\', \'k600\' or \'ssv2\'.'''							)
				_lowerCamelCase					     = json.load(open(hf_hub_download(lowercase_  ,      lowercase_  ,      repo_type='''dataset'''							)  ,      '''r'''							)							)
				_lowerCamelCase					     = {int(lowercase_							): v for k, v in idalabel.items()}
				_lowerCamelCase					     = idalabel
				_lowerCamelCase					     = {v: k for k, v in idalabel.items()}

				return config



def 				lowerCAmelCase_(      lowercase_   :							Optional[Any]							)		->				str:
				if "encoder." in name:
								_lowerCamelCase					     = name.replace('''encoder.'''  ,      ''''''							)
				if "cls_token" in name:
								_lowerCamelCase					     = name.replace('''cls_token'''  ,      '''timesformer.embeddings.cls_token'''							)
				if "pos_embed" in name:
								_lowerCamelCase					     = name.replace('''pos_embed'''  ,      '''timesformer.embeddings.position_embeddings'''							)
				if "time_embed" in name:
								_lowerCamelCase					     = name.replace('''time_embed'''  ,      '''timesformer.embeddings.time_embeddings'''							)
				if "patch_embed.proj" in name:
								_lowerCamelCase					     = name.replace('''patch_embed.proj'''  ,      '''timesformer.embeddings.patch_embeddings.projection'''							)
				if "patch_embed.norm" in name:
								_lowerCamelCase					     = name.replace('''patch_embed.norm'''  ,      '''timesformer.embeddings.norm'''							)
				if "blocks" in name:
								_lowerCamelCase					     = name.replace('''blocks'''  ,      '''timesformer.encoder.layer'''							)
				if "attn.proj" in name:
								_lowerCamelCase					     = name.replace('''attn.proj'''  ,      '''attention.output.dense'''							)
				if "attn" in name and "bias" not in name and "temporal" not in name:
								_lowerCamelCase					     = name.replace('''attn'''  ,      '''attention.self'''							)
				if "attn" in name and "temporal" not in name:
								_lowerCamelCase					     = name.replace('''attn'''  ,      '''attention.attention'''							)
				if "temporal_norm1" in name:
								_lowerCamelCase					     = name.replace('''temporal_norm1'''  ,      '''temporal_layernorm'''							)
				if "temporal_attn.proj" in name:
								_lowerCamelCase					     = name.replace('''temporal_attn'''  ,      '''temporal_attention.output.dense'''							)
				if "temporal_fc" in name:
								_lowerCamelCase					     = name.replace('''temporal_fc'''  ,      '''temporal_dense'''							)
				if "norm1" in name and "temporal" not in name:
								_lowerCamelCase					     = name.replace('''norm1'''  ,      '''layernorm_before'''							)
				if "norm2" in name:
								_lowerCamelCase					     = name.replace('''norm2'''  ,      '''layernorm_after'''							)
				if "mlp.fc1" in name:
								_lowerCamelCase					     = name.replace('''mlp.fc1'''  ,      '''intermediate.dense'''							)
				if "mlp.fc2" in name:
								_lowerCamelCase					     = name.replace('''mlp.fc2'''  ,      '''output.dense'''							)
				if "norm.weight" in name and "fc" not in name and "temporal" not in name:
								_lowerCamelCase					     = name.replace('''norm.weight'''  ,      '''timesformer.layernorm.weight'''							)
				if "norm.bias" in name and "fc" not in name and "temporal" not in name:
								_lowerCamelCase					     = name.replace('''norm.bias'''  ,      '''timesformer.layernorm.bias'''							)
				if "head" in name:
								_lowerCamelCase					     = name.replace('''head'''  ,      '''classifier'''							)

				return name



def 				lowerCAmelCase_(      lowercase_   :							str  ,      lowercase_   :							Optional[Any]							)		->				Optional[Any]:
				for key in orig_state_dict.copy().keys():
								_lowerCamelCase					     = orig_state_dict.pop(lowercase_							)

								if key.startswith('''model.'''							):
												_lowerCamelCase					     = key.replace('''model.'''  ,      ''''''							)

								if "qkv" in key:
												_lowerCamelCase					     = key.split('''.'''							)
												_lowerCamelCase					     = int(key_split[1]							)
												_lowerCamelCase					     = '''timesformer.encoder.layer.'''
												if "temporal" in key:
																_lowerCamelCase					     = '''.temporal_attention.attention.qkv.'''
												else:
																_lowerCamelCase					     = '''.attention.attention.qkv.'''
												if "weight" in key:
																_lowerCamelCase					     = val
												else:
																_lowerCamelCase					     = val
								else:
												_lowerCamelCase					     = val

				return orig_state_dict



def 				lowerCAmelCase_(      )		->				str:
				_lowerCamelCase					     = hf_hub_download(
				    repo_id='''hf-internal-testing/spaghetti-video'''  ,      filename='''eating_spaghetti.npy'''  ,      repo_type='''dataset'''							)
				_lowerCamelCase					     = np.load(lowercase_							)
				return list(lowercase_							)



def 				lowerCAmelCase_(      lowercase_   :							Any  ,      lowercase_   :							Union[str, Any]  ,      lowercase_   :							Any  ,      lowercase_   :							Optional[int]							)		->				int:
				_lowerCamelCase					     = get_timesformer_config(lowercase_							)

				_lowerCamelCase					     = TimesformerForVideoClassification(lowercase_							)

				# download original checkpoint, hosted on Google Drive
				_lowerCamelCase					     = '''pytorch_model.bin'''
				gdown.cached_download(lowercase_  ,      lowercase_  ,      quiet=lowercase_							)
				_lowerCamelCase					     = torch.load(lowercase_  ,      map_location='''cpu'''							)
				if "model" in files:
								_lowerCamelCase					     = files['''model''']
				elif "module" in files:
								_lowerCamelCase					     = files['''module''']
				else:
								_lowerCamelCase					     = files['''model_state''']
				_lowerCamelCase					     = convert_state_dict(lowercase_  ,      lowercase_							)

				model.load_state_dict(lowercase_							)
				model.eval()

				# verify model on basic input
				_lowerCamelCase					     = VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]  ,      image_std=[0.5, 0.5, 0.5]							)
				_lowerCamelCase					     = prepare_video()
				_lowerCamelCase					     = image_processor(video[:8]  ,      return_tensors='''pt'''							)

				_lowerCamelCase					     = model(**lowercase_							)
				_lowerCamelCase					     = outputs.logits

				_lowerCamelCase					     = [
				    # Kinetics-400 checkpoints (hr = high resolution input of 448px instead of 224px)
				    '''timesformer-base-finetuned-k400''',
				    '''timesformer-large-finetuned-k400''',
				    '''timesformer-hr-finetuned-k400''',
				    # Kinetics-600 checkpoints (hr = high resolution input of 448px instead of 224px)
				    '''timesformer-base-finetuned-k600''',
				    '''timesformer-large-finetuned-k600''',
				    '''timesformer-hr-finetuned-k600''',
				    # Something-Something-v2 checkpoints (hr = high resolution input of 448px instead of 224px)
				    '''timesformer-base-finetuned-ssv2''',
				    '''timesformer-large-finetuned-ssv2''',
				    '''timesformer-hr-finetuned-ssv2''',
				]

				# NOTE: logits were tested with image_mean and image_std equal to [0.5, 0.5, 0.5] and [0.5, 0.5, 0.5]
				if model_name == "timesformer-base-finetuned-k400":
								_lowerCamelCase					     = torch.Size([1, 4_00]							)
								_lowerCamelCase					     = torch.tensor([-0.3_0_1_6, -0.7_7_1_3, -0.4_2_0_5]							)
				elif model_name == "timesformer-base-finetuned-k600":
								_lowerCamelCase					     = torch.Size([1, 6_00]							)
								_lowerCamelCase					     = torch.tensor([-0.7_2_6_7, -0.7_4_6_6, 3.2_4_0_4]							)
				elif model_name == "timesformer-base-finetuned-ssv2":
								_lowerCamelCase					     = torch.Size([1, 1_74]							)
								_lowerCamelCase					     = torch.tensor([-0.9_0_5_9, 0.6_4_3_3, -3.1_4_5_7]							)
				elif model_name == "timesformer-large-finetuned-k400":
								_lowerCamelCase					     = torch.Size([1, 4_00]							)
								_lowerCamelCase					     = torch.tensor([0, 0, 0]							)
				elif model_name == "timesformer-large-finetuned-k600":
								_lowerCamelCase					     = torch.Size([1, 6_00]							)
								_lowerCamelCase					     = torch.tensor([0, 0, 0]							)
				elif model_name == "timesformer-large-finetuned-ssv2":
								_lowerCamelCase					     = torch.Size([1, 1_74]							)
								_lowerCamelCase					     = torch.tensor([0, 0, 0]							)
				elif model_name == "timesformer-hr-finetuned-k400":
								_lowerCamelCase					     = torch.Size([1, 4_00]							)
								_lowerCamelCase					     = torch.tensor([-0.9_6_1_7, -3.7_3_1_1, -3.7_7_0_8]							)
				elif model_name == "timesformer-hr-finetuned-k600":
								_lowerCamelCase					     = torch.Size([1, 6_00]							)
								_lowerCamelCase					     = torch.tensor([2.5_2_7_3, 0.7_1_2_7, 1.8_8_4_8]							)
				elif model_name == "timesformer-hr-finetuned-ssv2":
								_lowerCamelCase					     = torch.Size([1, 1_74]							)
								_lowerCamelCase					     = torch.tensor([-3.6_7_5_6, -0.7_5_1_3, 0.7_1_8_0]							)
				else:
								raise ValueError(F"""Model name not supported. Should be one of {model_names}"""							)

				# verify logits
				assert logits.shape == expected_shape
				assert torch.allclose(logits[0, :3]  ,      lowercase_  ,      atol=1e-4							)
				print('''Logits ok!'''							)

				if pytorch_dump_folder_path is not None:
								print(F"""Saving model and image processor to {pytorch_dump_folder_path}"""							)
								image_processor.save_pretrained(lowercase_							)
								model.save_pretrained(lowercase_							)

				if push_to_hub:
								print('''Pushing to the hub...'''							)
								model.push_to_hub(F"""fcakyon/{model_name}"""							)


if __name__ == "__main__":
			__SCREAMING_SNAKE_CASE     :  Any										=					argparse.ArgumentParser()
			# Required parameters
			parser.add_argument(
			    '''--checkpoint_url''',
			    default='''https://drive.google.com/u/1/uc?id=17yvuYp9L4mn-HpIcK5Zo6K3UoOy1kA5l&export=download''',
			    type=str,
			    help=(
			        '''URL of the original PyTorch checkpoint (on Google Drive) you\'d like to convert. Should be a direct'''
			        ''' download link.'''
			    ),
			)
			parser.add_argument(
			    '''--pytorch_dump_folder_path''',
			    default='''''',
			    type=str,
			    help='''Path to the output PyTorch model directory.''',
			)
			parser.add_argument('''--model_name''', default='''timesformer-base-finetuned-k400''', type=str, help='''Name of the model.''')
			parser.add_argument(
			    '''--push_to_hub''', action='''store_true''', help='''Whether or not to push the converted model to the 🤗 hub.'''
			)

			__SCREAMING_SNAKE_CASE     :  Dict										=					parser.parse_args()
			convert_timesformer_checkpoint(
			    args.checkpoint_url, args.pytorch_dump_folder_path, args.model_name, args.push_to_hub
			)





"""simple docstring"""
import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
			import torch

if is_vision_available():
			from PIL import Image

			from transformers import BlipImageProcessor


class 							lowerCamelCase_(      unittest.TestCase			):





					'''simple docstring'''



					def __init__(			self ,				lowerCamelCase__ ,				lowerCamelCase__=7 ,				lowerCamelCase__=3 ,				lowerCamelCase__=1_8 ,				lowerCamelCase__=3_0 ,				lowerCamelCase__=4_0_0 ,				lowerCamelCase__=True ,				lowerCamelCase__=None ,				lowerCamelCase__=True ,				lowerCamelCase__=False ,				lowerCamelCase__=[0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3] ,				lowerCamelCase__=[0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1] ,				lowerCamelCase__=True ,				):
									_lowerCamelCase					     = size if size is not None else {'''height''': 2_0, '''width''': 2_0}
									_lowerCamelCase					     = parent
									_lowerCamelCase					     = batch_size
									_lowerCamelCase					     = num_channels
									_lowerCamelCase					     = image_size
									_lowerCamelCase					     = min_resolution
									_lowerCamelCase					     = max_resolution
									_lowerCamelCase					     = do_resize
									_lowerCamelCase					     = size
									_lowerCamelCase					     = do_normalize
									_lowerCamelCase					     = image_mean
									_lowerCamelCase					     = image_std
									_lowerCamelCase					     = do_pad
									_lowerCamelCase					     = do_convert_rgb

					def 	snake_case__					(			self      ):
									return {
									    "do_resize": self.do_resize,
									    "size": self.size,
									    "do_normalize": self.do_normalize,
									    "image_mean": self.image_mean,
									    "image_std": self.image_std,
									    "do_convert_rgb": self.do_convert_rgb,
									    "do_pad": self.do_pad,
									}





					def 	snake_case__					(			self ,				lowerCamelCase__=False ,				lowerCamelCase__=False ,				lowerCamelCase__=False      ):
									assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

									if equal_resolution:
													_lowerCamelCase					     = []
													for i in range(self.batch_size      ):
																	image_inputs.append(
																	    np.random.randint(
																	        2_5_5 ,				size=(self.num_channels, self.max_resolution, self.max_resolution) ,				dtype=np.uinta      )      )
									else:
													_lowerCamelCase					     = []
													for i in range(self.batch_size      ):
																	_lowerCamelCase     ,   _lowerCamelCase					     = np.random.choice(np.arange(self.min_resolution ,				self.max_resolution      ) ,				2      )
																	image_inputs.append(np.random.randint(2_5_5 ,				size=(self.num_channels, width, height) ,				dtype=np.uinta      )      )

									if not numpify and not torchify:
													# PIL expects the channel dimension as last dimension
													_lowerCamelCase					     = [Image.fromarray(np.moveaxis(lowerCamelCase__ ,				0 ,				-1      )      ) for x in image_inputs]

									if torchify:
													_lowerCamelCase					     = [torch.from_numpy(lowerCamelCase__      ) for x in image_inputs]

									return image_inputs


@require_torch
@require_vision
class 							lowerCamelCase_(      A__, unittest.TestCase			):





					'''simple docstring'''



					lowercase__      :				Any        =       BlipImageProcessor if is_vision_available() else None

					def 	snake_case__					(			self      ):
									_lowerCamelCase					     = BlipImageProcessingTester(self      )

					@property
					def 	snake_case__					(			self      ):
									return self.image_processor_tester.prepare_image_processor_dict()

					def 	snake_case__					(			self      ):
									_lowerCamelCase					     = self.image_processing_class(**self.image_processor_dict      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''do_resize'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''size'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''do_normalize'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''image_mean'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''image_std'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''do_convert_rgb'''      )      )

					def 	snake_case__					(			self      ):
									pass

					def 	snake_case__					(			self      ):
									# Initialize image_processor
									_lowerCamelCase					     = self.image_processing_class(**self.image_processor_dict      )
									# create random PIL images
									_lowerCamelCase					     = self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__      )
									for image in image_inputs:
													self.assertIsInstance(lowerCamelCase__ ,				Image.Image      )

									# Test not batched input
									_lowerCamelCase					     = image_processor(image_inputs[0] ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)

									# Test batched
									_lowerCamelCase					     = image_processor(lowerCamelCase__ ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)

					def 	snake_case__					(			self      ):
									# Initialize image_processor
									_lowerCamelCase					     = self.image_processing_class(**self.image_processor_dict      )
									# create random numpy tensors
									_lowerCamelCase					     = self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__ ,				numpify=lowerCamelCase__      )
									for image in image_inputs:
													self.assertIsInstance(lowerCamelCase__ ,				np.ndarray      )

									# Test not batched input
									_lowerCamelCase					     = image_processor(image_inputs[0] ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)

									# Test batched
									_lowerCamelCase					     = image_processor(lowerCamelCase__ ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)





					def 	snake_case__					(			self      ):
									# Initialize image_processor
									_lowerCamelCase					     = self.image_processing_class(**self.image_processor_dict      )
									# create random PyTorch tensors
									_lowerCamelCase					     = self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__ ,				torchify=lowerCamelCase__      )
									for image in image_inputs:
													self.assertIsInstance(lowerCamelCase__ ,				torch.Tensor      )

									# Test not batched input
									_lowerCamelCase					     = image_processor(image_inputs[0] ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)

									# Test batched
									_lowerCamelCase					     = image_processor(lowerCamelCase__ ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)


@require_torch
@require_vision
class 							lowerCamelCase_(      A__, unittest.TestCase			):





					'''simple docstring'''



					lowercase__      :				Tuple        =       BlipImageProcessor if is_vision_available() else None

					def 	snake_case__					(			self      ):
									_lowerCamelCase					     = BlipImageProcessingTester(self ,				num_channels=4      )
									_lowerCamelCase					     = 3

					@property
					def 	snake_case__					(			self      ):
									return self.image_processor_tester.prepare_image_processor_dict()

					def 	snake_case__					(			self      ):
									_lowerCamelCase					     = self.image_processing_class(**self.image_processor_dict      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''do_resize'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''size'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''do_normalize'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''image_mean'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''image_std'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''do_convert_rgb'''      )      )

					def 	snake_case__					(			self      ):
									pass





					def 	snake_case__					(			self      ):
									# Initialize image_processor
									_lowerCamelCase					     = self.image_processing_class(**self.image_processor_dict      )
									# create random PIL images
									_lowerCamelCase					     = self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__      )
									for image in image_inputs:
													self.assertIsInstance(lowerCamelCase__ ,				Image.Image      )

									# Test not batched input
									_lowerCamelCase					     = image_processor(image_inputs[0] ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        1,
									        self.expected_encoded_image_num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)

									# Test batched
									_lowerCamelCase					     = image_processor(lowerCamelCase__ ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        self.image_processor_tester.batch_size,
									        self.expected_encoded_image_num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)
