



'''simple docstring'''





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 snake_case__  (			__SCREAMING_SNAKE_CASE )		->							Tuple:
		UpperCAmelCase_						=				VideoMAEConfig()

		set_architecture_configs(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE )

		if "finetuned" not in model_name:
				UpperCAmelCase_						=				False

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

		return config







def snake_case__  (			__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE )		->							int:







		if "small" in model_name:
				UpperCAmelCase_						=				384
				UpperCAmelCase_						=				1536
				UpperCAmelCase_						=				12
				UpperCAmelCase_						=				16
				UpperCAmelCase_						=				12
				UpperCAmelCase_						=				3
				UpperCAmelCase_						=				192
				UpperCAmelCase_						=				768
		elif "large" in model_name:
				UpperCAmelCase_						=				1024
				UpperCAmelCase_						=				4096
				UpperCAmelCase_						=				24
				UpperCAmelCase_						=				16
				UpperCAmelCase_						=				12
				UpperCAmelCase_						=				8
				UpperCAmelCase_						=				512
				UpperCAmelCase_						=				2048
		elif "huge" in model_name:
				UpperCAmelCase_						=				1280
				UpperCAmelCase_						=				5120
				UpperCAmelCase_						=				32
				UpperCAmelCase_						=				16
				UpperCAmelCase_						=				12
				UpperCAmelCase_						=				8
				UpperCAmelCase_						=				640
				UpperCAmelCase_						=				2560
		elif "base" not in model_name:
				raise ValueError("Model name should include either \"small\", \"base\", \"large\", or \"huge\"" )







def snake_case__  (			__SCREAMING_SNAKE_CASE )		->							List[str]:







		if "encoder." in name:
				UpperCAmelCase_						=				name.replace("encoder."			,     "" )
		if "cls_token" in name:
				UpperCAmelCase_						=				name.replace("cls_token"			,     "videomae.embeddings.cls_token" )
		if "decoder_pos_embed" in name:
				UpperCAmelCase_						=				name.replace("decoder_pos_embed"			,     "decoder.decoder_pos_embed" )
		if "pos_embed" in name and "decoder" not in name:
				UpperCAmelCase_						=				name.replace("pos_embed"			,     "videomae.embeddings.position_embeddings" )
		if "patch_embed.proj" in name:
				UpperCAmelCase_						=				name.replace("patch_embed.proj"			,     "videomae.embeddings.patch_embeddings.projection" )
		if "patch_embed.norm" in name:
				UpperCAmelCase_						=				name.replace("patch_embed.norm"			,     "videomae.embeddings.norm" )
		if "decoder.blocks" in name:
				UpperCAmelCase_						=				name.replace("decoder.blocks"			,     "decoder.decoder_layers" )
		if "blocks" in name:
				UpperCAmelCase_						=				name.replace("blocks"			,     "videomae.encoder.layer" )
		if "attn.proj" in name:
				UpperCAmelCase_						=				name.replace("attn.proj"			,     "attention.output.dense" )
		if "attn" in name and "bias" not in name:
				UpperCAmelCase_						=				name.replace("attn"			,     "attention.self" )
		if "attn" in name:
				UpperCAmelCase_						=				name.replace("attn"			,     "attention.attention" )
		if "norm1" in name:
				UpperCAmelCase_						=				name.replace("norm1"			,     "layernorm_before" )
		if "norm2" in name:
				UpperCAmelCase_						=				name.replace("norm2"			,     "layernorm_after" )
		if "mlp.fc1" in name:
				UpperCAmelCase_						=				name.replace("mlp.fc1"			,     "intermediate.dense" )
		if "mlp.fc2" in name:
				UpperCAmelCase_						=				name.replace("mlp.fc2"			,     "output.dense" )
		if "decoder_embed" in name:
				UpperCAmelCase_						=				name.replace("decoder_embed"			,     "decoder.decoder_embed" )
		if "decoder_norm" in name:
				UpperCAmelCase_						=				name.replace("decoder_norm"			,     "decoder.decoder_norm" )
		if "decoder_pred" in name:
				UpperCAmelCase_						=				name.replace("decoder_pred"			,     "decoder.decoder_pred" )
		if "norm.weight" in name and "decoder" not in name and "fc" not in name:
				UpperCAmelCase_						=				name.replace("norm.weight"			,     "videomae.layernorm.weight" )
		if "norm.bias" in name and "decoder" not in name and "fc" not in name:
				UpperCAmelCase_						=				name.replace("norm.bias"			,     "videomae.layernorm.bias" )
		if "head" in name and "decoder" not in name:
				UpperCAmelCase_						=				name.replace("head"			,     "classifier" )

		return name







def snake_case__  (			__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE )		->							List[Any]:







		for key in orig_state_dict.copy().keys():
				UpperCAmelCase_						=				orig_state_dict.pop(__SCREAMING_SNAKE_CASE )

				if key.startswith("encoder." ):
						UpperCAmelCase_						=				key.replace("encoder."			,     "" )

				if "qkv" in key:
						UpperCAmelCase_						=				key.split("." )
						if key.startswith("decoder.blocks" ):
								UpperCAmelCase_						=				config.decoder_hidden_size
								UpperCAmelCase_						=				int(key_split[2] )
								UpperCAmelCase_						=				"""decoder.decoder_layers."""
								if "weight" in key:
										UpperCAmelCase_						=				val[:dim, :]
										UpperCAmelCase_						=				val[dim : dim * 2, :]
										UpperCAmelCase_						=				val[-dim:, :]
						else:
								UpperCAmelCase_						=				config.hidden_size
								UpperCAmelCase_						=				int(key_split[1] )
								UpperCAmelCase_						=				"""videomae.encoder.layer."""
								if "weight" in key:
										UpperCAmelCase_						=				val[:dim, :]
										UpperCAmelCase_						=				val[dim : dim * 2, :]
										UpperCAmelCase_						=				val[-dim:, :]
				else:
						UpperCAmelCase_						=				val

		return orig_state_dict







def snake_case__  (			)		->							Union[str, Any]:
		UpperCAmelCase_						=				hf_hub_download(
		    repo_id="hf-internal-testing/spaghetti-video"			,     filename="eating_spaghetti.npy"			,     repo_type="dataset" )
		UpperCAmelCase_						=				np.load(__SCREAMING_SNAKE_CASE )
		return list(__SCREAMING_SNAKE_CASE )







def snake_case__  (			__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE )		->							Any:
		UpperCAmelCase_						=				get_videomae_config(__SCREAMING_SNAKE_CASE )

		if "finetuned" in model_name:
				UpperCAmelCase_						=				VideoMAEForVideoClassification(__SCREAMING_SNAKE_CASE )
		else:
				UpperCAmelCase_						=				VideoMAEForPreTraining(__SCREAMING_SNAKE_CASE )

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

		model.load_state_dict(__SCREAMING_SNAKE_CASE )
		model.eval()

		# verify model on basic input
		UpperCAmelCase_						=				VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]			,     image_std=[0.5, 0.5, 0.5] )
		UpperCAmelCase_						=				prepare_video()
		UpperCAmelCase_						=				image_processor(__SCREAMING_SNAKE_CASE			,     return_tensors="pt" )

		if "finetuned" not in model_name:
				UpperCAmelCase_						=				hf_hub_download(repo_id="hf-internal-testing/bool-masked-pos"			,     filename="bool_masked_pos.pt" )
				UpperCAmelCase_						=				torch.load(__SCREAMING_SNAKE_CASE )

		UpperCAmelCase_						=				model(**__SCREAMING_SNAKE_CASE )
		UpperCAmelCase_						=				outputs.logits

		UpperCAmelCase_						=				[
		    """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_						=				torch.Size([1, 400] )
				UpperCAmelCase_						=				torch.tensor([-0.9_291, -0.4_061, -0.9_307] )
		elif model_name == "videomae-small-finetuned-ssv2":
				UpperCAmelCase_						=				torch.Size([1, 174] )
				UpperCAmelCase_						=				torch.tensor([0.2_671, -0.4_689, -0.8_235] )
		elif model_name == "videomae-base":
				UpperCAmelCase_						=				torch.Size([1, 1408, 1536] )
				UpperCAmelCase_						=				torch.tensor([[0.7_739, 0.7_968, 0.7_089], [0.6_701, 0.7_487, 0.6_209], [0.4_287, 0.5_158, 0.4_773]] )
		elif model_name == "videomae-base-short":
				UpperCAmelCase_						=				torch.Size([1, 1408, 1536] )
				UpperCAmelCase_						=				torch.tensor([[0.7_994, 0.9_612, 0.8_508], [0.7_401, 0.8_958, 0.8_302], [0.5_862, 0.7_468, 0.7_325]] )
				# we verified the loss both for normalized and unnormalized targets for this one
				UpperCAmelCase_						=				torch.tensor([0.5_142] ) if config.norm_pix_loss else torch.tensor([0.6_469] )
		elif model_name == "videomae-large":
				UpperCAmelCase_						=				torch.Size([1, 1408, 1536] )
				UpperCAmelCase_						=				torch.tensor([[0.7_149, 0.7_997, 0.6_966], [0.6_768, 0.7_869, 0.6_948], [0.5_139, 0.6_221, 0.5_605]] )
		elif model_name == "videomae-large-finetuned-kinetics":
				UpperCAmelCase_						=				torch.Size([1, 400] )
				UpperCAmelCase_						=				torch.tensor([0.0_771, 0.0_011, -0.3_625] )
		elif model_name == "videomae-huge-finetuned-kinetics":
				UpperCAmelCase_						=				torch.Size([1, 400] )
				UpperCAmelCase_						=				torch.tensor([0.2_433, 0.1_632, -0.4_894] )
		elif model_name == "videomae-base-short-finetuned-kinetics":
				UpperCAmelCase_						=				torch.Size([1, 400] )
				UpperCAmelCase_						=				torch.tensor([0.6_588, 0.0_990, -0.2_493] )
		elif model_name == "videomae-base-finetuned-kinetics":
				UpperCAmelCase_						=				torch.Size([1, 400] )
				UpperCAmelCase_						=				torch.tensor([0.3_669, -0.0_688, -0.2_421] )
		elif model_name == "videomae-base-short-ssv2":
				UpperCAmelCase_						=				torch.Size([1, 1408, 1536] )
				UpperCAmelCase_						=				torch.tensor([[0.4_712, 0.5_296, 0.5_786], [0.2_278, 0.2_729, 0.4_026], [0.0_352, 0.0_730, 0.2_506]] )
		elif model_name == "videomae-base-short-finetuned-ssv2":
				UpperCAmelCase_						=				torch.Size([1, 174] )
				UpperCAmelCase_						=				torch.tensor([-0.0_537, -0.1_539, -0.3_266] )
		elif model_name == "videomae-base-ssv2":
				UpperCAmelCase_						=				torch.Size([1, 1408, 1536] )
				UpperCAmelCase_						=				torch.tensor([[0.8_131, 0.8_727, 0.8_546], [0.7_366, 0.9_377, 0.8_870], [0.5_935, 0.8_874, 0.8_564]] )
		elif model_name == "videomae-base-finetuned-ssv2":
				UpperCAmelCase_						=				torch.Size([1, 174] )
				UpperCAmelCase_						=				torch.tensor([0.1_961, -0.8_337, -0.6_389] )
		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]			,     __SCREAMING_SNAKE_CASE			,     atol=1E-4 )
		else:
				print("Logits:"			,     logits[0, :3, :3] )
				assert torch.allclose(logits[0, :3, :3]			,     __SCREAMING_SNAKE_CASE			,     atol=1E-4 )
		print("Logits ok!" )

		# verify loss, if applicable
		if model_name == "videomae-base-short":
				UpperCAmelCase_						=				outputs.loss
				assert torch.allclose(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE			,     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(__SCREAMING_SNAKE_CASE )
				model.save_pretrained(__SCREAMING_SNAKE_CASE )

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


if __name__ == "__main__":
			SCREAMING_SNAKE_CASE     				=						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."
			)

			SCREAMING_SNAKE_CASE     				=						parser.parse_args()
			convert_videomae_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path, args.model_name, args.push_to_hub)






from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
			from PIL import Image

			from ..image_utils import load_image

if is_tf_available():
			from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_VISION_2_SEQ_MAPPING

if is_torch_available():
			import torch

			from ..models.auto.modeling_auto import MODEL_FOR_VISION_2_SEQ_MAPPING

SCREAMING_SNAKE_CASE     				=						logging.get_logger(__name__)



@add_end_docstrings(lowercase__  )
class 	lowerCamelCase      (						lowercase__  ):







					'''simple docstring'''







					def __init__(			self    ,	*lowerCAmelCase    ,	**lowerCAmelCase     ):
							super().__init__(*lowerCAmelCase    ,	**lowerCAmelCase     )
							requires_backends(self    ,	"vision"     )
							self.check_model_type(
							    TF_MODEL_FOR_VISION_2_SEQ_MAPPING if self.framework == "tf" else MODEL_FOR_VISION_2_SEQ_MAPPING     )



					def    A__		(			self    ,	lowerCAmelCase=None    ,	lowerCAmelCase=None    ,	lowerCAmelCase=None     ):
							UpperCAmelCase_						=				{}
							UpperCAmelCase_						=				{}

							if prompt is not None:
									UpperCAmelCase_						=				prompt

							if generate_kwargs is not None:
									UpperCAmelCase_						=				generate_kwargs
							if max_new_tokens is not None:
									if "generate_kwargs" not in forward_kwargs:
											UpperCAmelCase_						=				{}
									if "max_new_tokens" in forward_kwargs["generate_kwargs"]:
											raise ValueError(
											    "'max_new_tokens' is defined twice, once in 'generate_kwargs' and once as a direct parameter,"
											    " please use only one"     )
									UpperCAmelCase_						=				max_new_tokens
							return preprocess_params, forward_kwargs, {}



					def __call__(			self    ,	lowerCAmelCase    ,	**lowerCAmelCase     ):
							return super().__call__(lowerCAmelCase    ,	**lowerCAmelCase     )



					def    A__		(			self    ,	lowerCAmelCase    ,	lowerCAmelCase=None     ):
							UpperCAmelCase_						=				load_image(lowerCAmelCase     )

							if prompt is not None:
									if not isinstance(lowerCAmelCase    ,	lowerCAmelCase     ):
											raise ValueError(
											    f'''Received an invalid text input, got - {type(lowerCAmelCase     )} - but expected a single string. '''
											    "Note also that one single text can be provided for conditional image to text generation."     )

									UpperCAmelCase_						=				self.model.config.model_type

									if model_type == "git":
											UpperCAmelCase_						=				self.image_processor(images=lowerCAmelCase    ,	return_tensors=self.framework     )
											UpperCAmelCase_						=				self.tokenizer(text=lowerCAmelCase    ,	add_special_tokens=lowerCAmelCase     ).input_ids
											UpperCAmelCase_						=				[self.tokenizer.cls_token_id] + input_ids
											UpperCAmelCase_						=				torch.tensor(lowerCAmelCase     ).unsqueeze(0     )
											model_inputs.update({"input_ids": input_ids}     )

									elif model_type == "pix2struct":
											UpperCAmelCase_						=				self.image_processor(images=lowerCAmelCase    ,	header_text=lowerCAmelCase    ,	return_tensors=self.framework     )

									elif model_type != "vision-encoder-decoder":
											# vision-encoder-decoder does not support conditional generation
											UpperCAmelCase_						=				self.image_processor(images=lowerCAmelCase    ,	return_tensors=self.framework     )
											UpperCAmelCase_						=				self.tokenizer(lowerCAmelCase    ,	return_tensors=self.framework     )
											model_inputs.update(lowerCAmelCase     )

									else:
											raise ValueError(f'''Model type {model_type} does not support conditional text generation'''     )

							else:
									UpperCAmelCase_						=				self.image_processor(images=lowerCAmelCase    ,	return_tensors=self.framework     )

							if self.model.config.model_type == "git" and prompt is None:
									UpperCAmelCase_						=				None

							return model_inputs



					def    A__		(			self    ,	lowerCAmelCase    ,	lowerCAmelCase=None     ):
							# Git model sets `model_inputs["input_ids"] = None` in `preprocess` (when `prompt=None`). In batch model, the
							# pipeline will group them into a list of `None`, which fail `_forward`. Avoid this by checking it first.
							if (
							    "input_ids" in model_inputs
							    and isinstance(model_inputs["input_ids"]    ,	lowerCAmelCase     )
							    and all(x is None for x in model_inputs["input_ids"]     )
							):
									UpperCAmelCase_						=				None

							if generate_kwargs is None:
									UpperCAmelCase_						=				{}
							# FIXME: We need to pop here due to a difference in how `generation.py` and `generation.tf_utils.py`
							#  parse inputs. In the Tensorflow version, `generate` raises an error if we don't use `input_ids` whereas
							#  the PyTorch version matches it with `self.model.main_input_name` or `self.model.encoder.main_input_name`
							#  in the `_prepare_model_inputs` method.
							UpperCAmelCase_						=				model_inputs.pop(self.model.main_input_name     )
							UpperCAmelCase_						=				self.model.generate(lowerCAmelCase    ,	**lowerCAmelCase    ,	**lowerCAmelCase     )
							return model_outputs



					def    A__		(			self    ,	lowerCAmelCase     ):
							UpperCAmelCase_						=				[]
							for output_ids in model_outputs:
									UpperCAmelCase_						=				{
									    "generated_text": self.tokenizer.decode(
									        lowerCAmelCase    ,	skip_special_tokens=lowerCAmelCase    ,	)
									}
									records.append(lowerCAmelCase     )
							return records

