

import argparse
import json

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

from transformers import (
    VideoMAEConfig,
    VideoMAEForPreTraining,
    VideoMAEForVideoClassification,
    VideoMAEImageProcessor,
)

def       a__					(		UpperCAmelCase							:      Any      )  ->						int:
		UpperCAmelCase	:							Optional[Any]						=			VideoMAEConfig()

		set_architecture_configs(_A    ,			_A      )

		if "finetuned" not in model_name:
				UpperCAmelCase	:							Union[str, Any]						=			False

		if "finetuned" in model_name:
				UpperCAmelCase	:							Any						=			'huggingface/label-files'
				if "kinetics" in model_name:
						UpperCAmelCase	:							Dict						=			400
						UpperCAmelCase	:							Dict						=			'kinetics400-id2label.json'
				elif "ssv2" in model_name:
						UpperCAmelCase	:							Any						=			174
						UpperCAmelCase	:							Union[str, Any]						=			'something-something-v2-id2label.json'
				else:
						raise ValueError('''Model name should either contain \'kinetics\' or \'ssv2\' in case it\'s fine-tuned.'''      )
				UpperCAmelCase	:							Union[str, Any]						=			json.load(open(hf_hub_download(_A    ,			_A    ,			repo_type='''dataset'''      )    ,			'''r'''      )      )
				UpperCAmelCase	:							str						=			{int(_A      ): v for k, v in idalabel.items()}
				UpperCAmelCase	:							str						=			idalabel
				UpperCAmelCase	:							Optional[Any]						=			{v: k for k, v in idalabel.items()}

		return config

def       a__					(		UpperCAmelCase							:      List[Any]    ,			UpperCAmelCase							:      Optional[int]      )  ->						Dict:
		if "small" in model_name:
				UpperCAmelCase	:							Any						=			384
				UpperCAmelCase	:							str						=			1_536
				UpperCAmelCase	:							Optional[Any]						=			12
				UpperCAmelCase	:							Union[str, Any]						=			16
				UpperCAmelCase	:							Optional[Any]						=			12
				UpperCAmelCase	:							Tuple						=			3
				UpperCAmelCase	:							str						=			192
				UpperCAmelCase	:							Dict						=			768
		elif "large" in model_name:
				UpperCAmelCase	:							Any						=			1_024
				UpperCAmelCase	:							Dict						=			4_096
				UpperCAmelCase	:							str						=			24
				UpperCAmelCase	:							str						=			16
				UpperCAmelCase	:							Union[str, Any]						=			12
				UpperCAmelCase	:							int						=			8
				UpperCAmelCase	:							List[str]						=			512
				UpperCAmelCase	:							Union[str, Any]						=			2_048
		elif "huge" in model_name:
				UpperCAmelCase	:							Union[str, Any]						=			1_280
				UpperCAmelCase	:							Dict						=			5_120
				UpperCAmelCase	:							Optional[int]						=			32
				UpperCAmelCase	:							Dict						=			16
				UpperCAmelCase	:							Dict						=			12
				UpperCAmelCase	:							List[str]						=			8
				UpperCAmelCase	:							List[str]						=			640
				UpperCAmelCase	:							Union[str, Any]						=			2_560
		elif "base" not in model_name:
				raise ValueError('''Model name should include either "small", "base", "large", or "huge"'''      )

def       a__					(		UpperCAmelCase							:      str      )  ->						Optional[int]:
		if "encoder." in name:
				UpperCAmelCase	:							List[str]						=			name.replace('''encoder.'''    ,			''''''      )
		if "cls_token" in name:
				UpperCAmelCase	:							int						=			name.replace('''cls_token'''    ,			'''videomae.embeddings.cls_token'''      )
		if "decoder_pos_embed" in name:
				UpperCAmelCase	:							List[Any]						=			name.replace('''decoder_pos_embed'''    ,			'''decoder.decoder_pos_embed'''      )
		if "pos_embed" in name and "decoder" not in name:
				UpperCAmelCase	:							int						=			name.replace('''pos_embed'''    ,			'''videomae.embeddings.position_embeddings'''      )
		if "patch_embed.proj" in name:
				UpperCAmelCase	:							Tuple						=			name.replace('''patch_embed.proj'''    ,			'''videomae.embeddings.patch_embeddings.projection'''      )
		if "patch_embed.norm" in name:
				UpperCAmelCase	:							Tuple						=			name.replace('''patch_embed.norm'''    ,			'''videomae.embeddings.norm'''      )
		if "decoder.blocks" in name:
				UpperCAmelCase	:							Optional[int]						=			name.replace('''decoder.blocks'''    ,			'''decoder.decoder_layers'''      )
		if "blocks" in name:
				UpperCAmelCase	:							Tuple						=			name.replace('''blocks'''    ,			'''videomae.encoder.layer'''      )
		if "attn.proj" in name:
				UpperCAmelCase	:							Optional[int]						=			name.replace('''attn.proj'''    ,			'''attention.output.dense'''      )
		if "attn" in name and "bias" not in name:
				UpperCAmelCase	:							Optional[int]						=			name.replace('''attn'''    ,			'''attention.self'''      )
		if "attn" in name:
				UpperCAmelCase	:							int						=			name.replace('''attn'''    ,			'''attention.attention'''      )
		if "norm1" in name:
				UpperCAmelCase	:							Union[str, Any]						=			name.replace('''norm1'''    ,			'''layernorm_before'''      )
		if "norm2" in name:
				UpperCAmelCase	:							Tuple						=			name.replace('''norm2'''    ,			'''layernorm_after'''      )
		if "mlp.fc1" in name:
				UpperCAmelCase	:							Tuple						=			name.replace('''mlp.fc1'''    ,			'''intermediate.dense'''      )
		if "mlp.fc2" in name:
				UpperCAmelCase	:							Any						=			name.replace('''mlp.fc2'''    ,			'''output.dense'''      )
		if "decoder_embed" in name:
				UpperCAmelCase	:							Tuple						=			name.replace('''decoder_embed'''    ,			'''decoder.decoder_embed'''      )
		if "decoder_norm" in name:
				UpperCAmelCase	:							Optional[Any]						=			name.replace('''decoder_norm'''    ,			'''decoder.decoder_norm'''      )
		if "decoder_pred" in name:
				UpperCAmelCase	:							List[Any]						=			name.replace('''decoder_pred'''    ,			'''decoder.decoder_pred'''      )
		if "norm.weight" in name and "decoder" not in name and "fc" not in name:
				UpperCAmelCase	:							Tuple						=			name.replace('''norm.weight'''    ,			'''videomae.layernorm.weight'''      )
		if "norm.bias" in name and "decoder" not in name and "fc" not in name:
				UpperCAmelCase	:							Optional[int]						=			name.replace('''norm.bias'''    ,			'''videomae.layernorm.bias'''      )
		if "head" in name and "decoder" not in name:
				UpperCAmelCase	:							Tuple						=			name.replace('''head'''    ,			'''classifier'''      )

		return name

def       a__					(		UpperCAmelCase							:      List[Any]    ,			UpperCAmelCase							:      List[Any]      )  ->						Any:
		for key in orig_state_dict.copy().keys():
				UpperCAmelCase	:							Any						=			orig_state_dict.pop(_A      )

				if key.startswith('''encoder.'''      ):
						UpperCAmelCase	:							Union[str, Any]						=			key.replace('''encoder.'''    ,			''''''      )

				if "qkv" in key:
						UpperCAmelCase	:							Dict						=			key.split('''.'''      )
						if key.startswith('''decoder.blocks'''      ):
								UpperCAmelCase	:							Union[str, Any]						=			config.decoder_hidden_size
								UpperCAmelCase	:							Any						=			int(key_split[2]      )
								UpperCAmelCase	:							Optional[Any]						=			'decoder.decoder_layers.'
								if "weight" in key:
										UpperCAmelCase	:							Dict						=			val[:dim, :]
										UpperCAmelCase	:							Dict						=			val[dim : dim * 2, :]
										UpperCAmelCase	:							Union[str, Any]						=			val[-dim:, :]
						else:
								UpperCAmelCase	:							Any						=			config.hidden_size
								UpperCAmelCase	:							int						=			int(key_split[1]      )
								UpperCAmelCase	:							Any						=			'videomae.encoder.layer.'
								if "weight" in key:
										UpperCAmelCase	:							Tuple						=			val[:dim, :]
										UpperCAmelCase	:							List[Any]						=			val[dim : dim * 2, :]
										UpperCAmelCase	:							Any						=			val[-dim:, :]
				else:
						UpperCAmelCase	:							Dict						=			val

		return orig_state_dict

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

def       a__					(		UpperCAmelCase							:      Union[str, Any]    ,			UpperCAmelCase							:      Optional[Any]    ,			UpperCAmelCase							:      Tuple    ,			UpperCAmelCase							:      List[str]      )  ->						int:
		UpperCAmelCase	:							Tuple						=			get_videomae_config(_A      )

		if "finetuned" in model_name:
				UpperCAmelCase	:							Union[str, Any]						=			VideoMAEForVideoClassification(_A      )
		else:
				UpperCAmelCase	:							List[Any]						=			VideoMAEForPreTraining(_A      )

		# download original checkpoint, hosted on Google Drive
		UpperCAmelCase	:							List[Any]						=			'pytorch_model.bin'
		gdown.cached_download(_A    ,			_A    ,			quiet=_A      )
		UpperCAmelCase	:							Dict						=			torch.load(_A    ,			map_location='''cpu'''      )
		if "model" in files:
				UpperCAmelCase	:							Dict						=			files['model']
		else:
				UpperCAmelCase	:							List[Any]						=			files['module']
		UpperCAmelCase	:							str						=			convert_state_dict(_A    ,			_A      )

		model.load_state_dict(_A      )
		model.eval()

		# verify model on basic input
		UpperCAmelCase	:							Dict						=			VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]    ,			image_std=[0.5, 0.5, 0.5]      )
		UpperCAmelCase	:							Optional[int]						=			prepare_video()
		UpperCAmelCase	:							Union[str, Any]						=			image_processor(_A    ,			return_tensors='''pt'''      )

		if "finetuned" not in model_name:
				UpperCAmelCase	:							Any						=			hf_hub_download(repo_id='''hf-internal-testing/bool-masked-pos'''    ,			filename='''bool_masked_pos.pt'''      )
				UpperCAmelCase	:							int						=			torch.load(_A      )

		UpperCAmelCase	:							str						=			model(**_A      )
		UpperCAmelCase	:							Optional[int]						=			outputs.logits

		UpperCAmelCase	:							List[Any]						=			[
		    'videomae-small-finetuned-kinetics',
		    'videomae-small-finetuned-ssv2',
		    # Kinetics-400 checkpoints (short = pretrained only for 800 epochs instead of 1600)
		    'videomae-base-short',
		    'videomae-base-short-finetuned-kinetics',
		    'videomae-base',
		    'videomae-base-finetuned-kinetics',
		    'videomae-large',
		    'videomae-large-finetuned-kinetics',
		    'videomae-huge-finetuned-kinetics',
		    # Something-Something-v2 checkpoints (short = pretrained only for 800 epochs instead of 2400)
		    'videomae-base-short-ssv2',
		    'videomae-base-short-finetuned-ssv2',
		    'videomae-base-ssv2',
		    'videomae-base-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 == "videomae-small-finetuned-kinetics":
				UpperCAmelCase	:							int						=			torch.Size([1, 400]      )
				UpperCAmelCase	:							str						=			torch.tensor([-0.9291, -0.4061, -0.9307]      )
		elif model_name == "videomae-small-finetuned-ssv2":
				UpperCAmelCase	:							Union[str, Any]						=			torch.Size([1, 174]      )
				UpperCAmelCase	:							List[Any]						=			torch.tensor([0.2671, -0.4689, -0.8235]      )
		elif model_name == "videomae-base":
				UpperCAmelCase	:							Optional[int]						=			torch.Size([1, 1_408, 1_536]      )
				UpperCAmelCase	:							Any						=			torch.tensor([[0.7739, 0.7968, 0.7089], [0.6701, 0.7487, 0.6209], [0.4287, 0.5158, 0.4773]]      )
		elif model_name == "videomae-base-short":
				UpperCAmelCase	:							str						=			torch.Size([1, 1_408, 1_536]      )
				UpperCAmelCase	:							Union[str, Any]						=			torch.tensor([[0.7994, 0.9612, 0.8508], [0.7401, 0.8958, 0.8302], [0.5862, 0.7468, 0.7325]]      )
				# we verified the loss both for normalized and unnormalized targets for this one
				UpperCAmelCase	:							int						=			torch.tensor([0.5142]      ) if config.norm_pix_loss else torch.tensor([0.6469]      )
		elif model_name == "videomae-large":
				UpperCAmelCase	:							Optional[Any]						=			torch.Size([1, 1_408, 1_536]      )
				UpperCAmelCase	:							Optional[Any]						=			torch.tensor([[0.7149, 0.7997, 0.6966], [0.6768, 0.7869, 0.6948], [0.5139, 0.6221, 0.5605]]      )
		elif model_name == "videomae-large-finetuned-kinetics":
				UpperCAmelCase	:							List[Any]						=			torch.Size([1, 400]      )
				UpperCAmelCase	:							Union[str, Any]						=			torch.tensor([0.0771, 0.0011, -0.3625]      )
		elif model_name == "videomae-huge-finetuned-kinetics":
				UpperCAmelCase	:							Optional[Any]						=			torch.Size([1, 400]      )
				UpperCAmelCase	:							List[Any]						=			torch.tensor([0.2433, 0.1632, -0.4894]      )
		elif model_name == "videomae-base-short-finetuned-kinetics":
				UpperCAmelCase	:							List[Any]						=			torch.Size([1, 400]      )
				UpperCAmelCase	:							Tuple						=			torch.tensor([0.6588, 0.0990, -0.2493]      )
		elif model_name == "videomae-base-finetuned-kinetics":
				UpperCAmelCase	:							Dict						=			torch.Size([1, 400]      )
				UpperCAmelCase	:							Optional[Any]						=			torch.tensor([0.3669, -0.0688, -0.2421]      )
		elif model_name == "videomae-base-short-ssv2":
				UpperCAmelCase	:							str						=			torch.Size([1, 1_408, 1_536]      )
				UpperCAmelCase	:							Any						=			torch.tensor([[0.4712, 0.5296, 0.5786], [0.2278, 0.2729, 0.4026], [0.0352, 0.0730, 0.2506]]      )
		elif model_name == "videomae-base-short-finetuned-ssv2":
				UpperCAmelCase	:							List[Any]						=			torch.Size([1, 174]      )
				UpperCAmelCase	:							int						=			torch.tensor([-0.0537, -0.1539, -0.3266]      )
		elif model_name == "videomae-base-ssv2":
				UpperCAmelCase	:							Optional[Any]						=			torch.Size([1, 1_408, 1_536]      )
				UpperCAmelCase	:							Tuple						=			torch.tensor([[0.8131, 0.8727, 0.8546], [0.7366, 0.9377, 0.8870], [0.5935, 0.8874, 0.8564]]      )
		elif model_name == "videomae-base-finetuned-ssv2":
				UpperCAmelCase	:							Tuple						=			torch.Size([1, 174]      )
				UpperCAmelCase	:							List[str]						=			torch.tensor([0.1961, -0.8337, -0.6389]      )
		else:
				raise ValueError(f'''Model name not supported. Should be one of {model_names}'''      )

		# verify logits
		assert logits.shape == expected_shape
		if "finetuned" in model_name:
				assert torch.allclose(logits[0, :3]    ,			_A    ,			atol=1E-4      )
		else:
				print('''Logits:'''    ,			logits[0, :3, :3]      )
				assert torch.allclose(logits[0, :3, :3]    ,			_A    ,			atol=1E-4      )
		print('''Logits ok!'''      )

		# verify loss, if applicable
		if model_name == "videomae-base-short":
				UpperCAmelCase	:							Dict						=			outputs.loss
				assert torch.allclose(_A    ,			_A    ,			atol=1E-4      )
				print('''Loss 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(_A      )
				model.save_pretrained(_A      )

		if push_to_hub:
				print('''Pushing to the hub...'''      )
				model.push_to_hub(_A    ,			organization='''nielsr'''      )


if __name__ == "__main__":
					_lowerCamelCase    :       str        =							argparse.ArgumentParser()
					# Required parameters
					parser.add_argument(
					    "--checkpoint_url",
					    default="https://drive.google.com/u/1/uc?id=1tEhLyskjb755TJ65ptsrafUG2llSwQE1&amp;export=download&amp;confirm=t&amp;uuid=aa3276eb-fb7e-482a-adec-dc7171df14c4",
					    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="/Users/nielsrogge/Documents/VideoMAE/Test",
					    type=str,
					    help="Path to the output PyTorch model directory.",
					)
					parser.add_argument("--model_name", default="videomae-base", 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."
					)

					_lowerCamelCase    :       List[str]        =							parser.parse_args()
					convert_videomae_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path, args.model_name, args.push_to_hub)








from typing import Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
     from PIL import Image

     from ..image_utils import load_image

if is_torch_available():
     from ..models.auto.modeling_auto import MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING

_lowerCamelCase    :       Dict        =							logging.get_logger(__name__)





@add_end_docstrings(lowerCamelCase__						)
class  __UpperCAmelCase					(	lowerCamelCase__						):



   def __init__(							self							:					Optional[Any],		*__A							:					Tuple,		**__A							:					Tuple						):
     super().__init__(*__A,		**__A						)
     self.check_model_type(__A						)



   def 	__magic_name__   (							self							:					Union[str, Any],		__A							:					int=None,		__A							:					Tuple=None,		__A							:					Any=None,		**__A							:					Optional[int]						):
     UpperCAmelCase			,						UpperCAmelCase	:							List[Any]						=			{}, {}
     if padding is not None:
       UpperCAmelCase	:							Optional[int]						=			padding
     if truncation is not None:
       UpperCAmelCase	:							Optional[int]						=			truncation
     if top_k is not None:
       UpperCAmelCase	:							Tuple						=			top_k
     return preprocess_params, {}, postprocess_params



   def __call__(							self							:					Union[str, Any],		__A							:					Union["Image.Image", str],		__A							:					str = None,		**__A							:					Optional[int]						):
     if isinstance(__A,		(Image.Image, str)						) and isinstance(__A,		__A						):
       UpperCAmelCase	:							int						=			{'''image''': image, '''question''': question}
     else:
       UpperCAmelCase	:							str						=			image
     UpperCAmelCase	:							Union[str, Any]						=			super().__call__(__A,		**__A						)
     return results



   def 	__magic_name__   (							self							:					List[str],		__A							:					Union[str, Any],		__A							:					Tuple=False,		__A							:					List[Any]=False						):
     UpperCAmelCase	:							int						=			load_image(inputs['''image''']						)
     UpperCAmelCase	:							List[str]						=			self.tokenizer(
         inputs['''question'''],		return_tensors=self.framework,		padding=__A,		truncation=__A						)
     UpperCAmelCase	:							Union[str, Any]						=			self.image_processor(images=__A,		return_tensors=self.framework						)
     model_inputs.update(__A						)
     return model_inputs



   def 	__magic_name__   (							self							:					Optional[Any],		__A							:					List[Any]						):
     UpperCAmelCase	:							Optional[int]						=			self.model(**__A						)
     return model_outputs





   def 	__magic_name__   (							self							:					Any,		__A							:					List[str],		__A							:					Union[str, Any]=5						):
     if top_k > self.model.config.num_labels:
       UpperCAmelCase	:							Any						=			self.model.config.num_labels

     if self.framework == "pt":
       UpperCAmelCase	:							Any						=			model_outputs.logits.sigmoid()[0]
       UpperCAmelCase			,						UpperCAmelCase	:							Union[str, Any]						=			probs.topk(__A						)
     else:
       raise ValueError(F'''Unsupported framework: {self.framework}'''						)

     UpperCAmelCase	:							str						=			scores.tolist()
     UpperCAmelCase	:							Tuple						=			ids.tolist()
     return [{"score": score, "answer": self.model.config.idalabel[_id]} for score, _id in zip(__A,		__A						)]





