





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      __lowercase     (   _SCREAMING_SNAKE_CASE			)				->     Union[str, Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE	=		VideoMAEConfig()

      set_architecture_configs(__SCREAMING_SNAKE_CASE				,		__SCREAMING_SNAKE_CASE			)

      if "finetuned" not in model_name:
            SCREAMING_SNAKE_CASE	=		False

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

      return config



def      __lowercase     (   _SCREAMING_SNAKE_CASE				,		_SCREAMING_SNAKE_CASE			)				->     Any:


      '''simple docstring'''


      if "small" in model_name:
            SCREAMING_SNAKE_CASE	=		3_84
            SCREAMING_SNAKE_CASE	=		15_36
            SCREAMING_SNAKE_CASE	=		12
            SCREAMING_SNAKE_CASE	=		16
            SCREAMING_SNAKE_CASE	=		12
            SCREAMING_SNAKE_CASE	=		3
            SCREAMING_SNAKE_CASE	=		1_92
            SCREAMING_SNAKE_CASE	=		7_68
      elif "large" in model_name:
            SCREAMING_SNAKE_CASE	=		10_24
            SCREAMING_SNAKE_CASE	=		40_96
            SCREAMING_SNAKE_CASE	=		24
            SCREAMING_SNAKE_CASE	=		16
            SCREAMING_SNAKE_CASE	=		12
            SCREAMING_SNAKE_CASE	=		8
            SCREAMING_SNAKE_CASE	=		5_12
            SCREAMING_SNAKE_CASE	=		20_48
      elif "huge" in model_name:
            SCREAMING_SNAKE_CASE	=		12_80
            SCREAMING_SNAKE_CASE	=		51_20
            SCREAMING_SNAKE_CASE	=		32
            SCREAMING_SNAKE_CASE	=		16
            SCREAMING_SNAKE_CASE	=		12
            SCREAMING_SNAKE_CASE	=		8
            SCREAMING_SNAKE_CASE	=		6_40
            SCREAMING_SNAKE_CASE	=		25_60
      elif "base" not in model_name:
            raise ValueError("""Model name should include either \"small\", \"base\", \"large\", or \"huge\""""			)



def      __lowercase     (   _SCREAMING_SNAKE_CASE			)				->     int:


      '''simple docstring'''


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

      return name



def      __lowercase     (   _SCREAMING_SNAKE_CASE				,		_SCREAMING_SNAKE_CASE			)				->     List[str]:


      '''simple docstring'''


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

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

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

      return orig_state_dict



def      __lowercase     (   )				->     List[str]:


      '''simple docstring'''


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



def      __lowercase     (   _SCREAMING_SNAKE_CASE				,		_SCREAMING_SNAKE_CASE				,		_SCREAMING_SNAKE_CASE				,		_SCREAMING_SNAKE_CASE			)				->     Optional[Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE	=		get_videomae_config(__SCREAMING_SNAKE_CASE			)

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

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

      model.load_state_dict(__SCREAMING_SNAKE_CASE			)
      model.eval()

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

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

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

      SCREAMING_SNAKE_CASE	=		[
          "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":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 4_00]			)
            SCREAMING_SNAKE_CASE	=		torch.tensor([-0.9_291, -0.4_061, -0.9_307]			)
      elif model_name == "videomae-small-finetuned-ssv2":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 1_74]			)
            SCREAMING_SNAKE_CASE	=		torch.tensor([0.2_671, -0.4_689, -0.8_235]			)
      elif model_name == "videomae-base":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 14_08, 15_36]			)
            SCREAMING_SNAKE_CASE	=		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":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 14_08, 15_36]			)
            SCREAMING_SNAKE_CASE	=		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
            SCREAMING_SNAKE_CASE	=		torch.tensor([0.5_142]			) if config.norm_pix_loss else torch.tensor([0.6_469]			)
      elif model_name == "videomae-large":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 14_08, 15_36]			)
            SCREAMING_SNAKE_CASE	=		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":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 4_00]			)
            SCREAMING_SNAKE_CASE	=		torch.tensor([0.0_771, 0.0_011, -0.3_625]			)
      elif model_name == "videomae-huge-finetuned-kinetics":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 4_00]			)
            SCREAMING_SNAKE_CASE	=		torch.tensor([0.2_433, 0.1_632, -0.4_894]			)
      elif model_name == "videomae-base-short-finetuned-kinetics":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 4_00]			)
            SCREAMING_SNAKE_CASE	=		torch.tensor([0.6_588, 0.0_990, -0.2_493]			)
      elif model_name == "videomae-base-finetuned-kinetics":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 4_00]			)
            SCREAMING_SNAKE_CASE	=		torch.tensor([0.3_669, -0.0_688, -0.2_421]			)
      elif model_name == "videomae-base-short-ssv2":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 14_08, 15_36]			)
            SCREAMING_SNAKE_CASE	=		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":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 1_74]			)
            SCREAMING_SNAKE_CASE	=		torch.tensor([-0.0_537, -0.1_539, -0.3_266]			)
      elif model_name == "videomae-base-ssv2":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 14_08, 15_36]			)
            SCREAMING_SNAKE_CASE	=		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":
            SCREAMING_SNAKE_CASE	=		torch.Size([1, 1_74]			)
            SCREAMING_SNAKE_CASE	=		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":
            SCREAMING_SNAKE_CASE	=		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)









# We ignore warnings about stepping the scheduler since we step it ourselves during gradient accumulation

import warnings

from .state import AcceleratorState, GradientState


warnings.filterwarnings("""ignore""", category=UserWarning, module="""torch.optim.lr_scheduler""")

class 	UpperCamelCase__				:



     '''simple docstring'''

     def __init__(							self		:					Optional[Any]		,lowerCamelCase__		:					Optional[int]		,lowerCamelCase__		:					Tuple		,lowerCamelCase__		:					bool = True		,lowerCamelCase__		:					bool = False )  ->	Tuple:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		scheduler
           SCREAMING_SNAKE_CASE	=		optimizers if isinstance(lowerCamelCase__		,(list, tuple) ) else [optimizers]
           SCREAMING_SNAKE_CASE	=		split_batches
           SCREAMING_SNAKE_CASE	=		step_with_optimizer
           SCREAMING_SNAKE_CASE	=		GradientState()






     def        SCREAMING_SNAKE_CASE__						(							self		:					str		,*lowerCamelCase__		:					Union[str, Any]		,**lowerCamelCase__		:					Optional[int] )  ->	Dict:




           '''simple docstring'''


           if not self.step_with_optimizer:
                 # No link between scheduler and optimizer -> just step
                 self.scheduler.step(*lowerCamelCase__		,**lowerCamelCase__ )
                 return

           # Otherwise, first make sure the optimizer was stepped.
           if not self.gradient_state.sync_gradients:
                 if self.gradient_state.adjust_scheduler:
                       self.scheduler._step_count += 1
                 return

           for opt in self.optimizers:
                 if opt.step_was_skipped:
                       return
           if self.split_batches:
                 # Split batches -> the training dataloader batch size is not changed so one step per training step
                 self.scheduler.step(*lowerCamelCase__		,**lowerCamelCase__ )
           else:
                 # Otherwise the training dataloader batch size was multiplied by `num_processes`, so we need to do
                 # num_processes steps per training step
                 SCREAMING_SNAKE_CASE	=		AcceleratorState().num_processes
                 for _ in range(lowerCamelCase__ ):
                       # Special case when using OneCycle and `drop_last` was not used
                       if hasattr(self.scheduler		,"""total_steps""" ):
                             if self.scheduler._step_count <= self.scheduler.total_steps:
                                   self.scheduler.step(*lowerCamelCase__		,**lowerCamelCase__ )
                       else:
                             self.scheduler.step(*lowerCamelCase__		,**lowerCamelCase__ )






     def        SCREAMING_SNAKE_CASE__						(							self		:					int )  ->	Optional[Any]:




           '''simple docstring'''


           return self.scheduler.get_last_lr()






     def        SCREAMING_SNAKE_CASE__						(							self		:					List[Any] )  ->	Optional[Any]:




           '''simple docstring'''


           return self.scheduler.state_dict()






     def        SCREAMING_SNAKE_CASE__						(							self		:					Optional[int]		,lowerCamelCase__		:					str )  ->	int:




           '''simple docstring'''


           self.scheduler.load_state_dict(lowerCamelCase__ )






     def        SCREAMING_SNAKE_CASE__						(							self		:					List[Any] )  ->	Optional[Any]:




           '''simple docstring'''


           return self.scheduler.get_lr()






     def        SCREAMING_SNAKE_CASE__						(							self		:					Dict		,*lowerCamelCase__		:					int		,**lowerCamelCase__		:					Tuple )  ->	Optional[int]:




           '''simple docstring'''


           return self.scheduler.print_lr(*lowerCamelCase__		,**lowerCamelCase__ )


