
'''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    (   A_							:     Dict ):
  """simple docstring"""







  a_   :						Any   =				VideoMAEConfig()

  set_architecture_configs(__lowerCamelCase  ,		__lowerCamelCase )

  if "finetuned" not in model_name:
    a_   :						int   =				False

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

  return config






def 						_snake_case    (   A_							:     Optional[Any]  ,		A_							:     int ):
  """simple docstring"""







  if "small" in model_name:
    a_   :						Dict   =				384
    a_   :						Optional[Any]   =				1536
    a_   :						Union[str, Any]   =				12
    a_   :						Dict   =				16
    a_   :						List[str]   =				12
    a_   :						List[str]   =				3
    a_   :						str   =				192
    a_   :						Union[str, Any]   =				768
  elif "large" in model_name:
    a_   :						Tuple   =				1024
    a_   :						List[Any]   =				4096
    a_   :						Any   =				24
    a_   :						Optional[int]   =				16
    a_   :						List[str]   =				12
    a_   :						int   =				8
    a_   :						Any   =				512
    a_   :						List[Any]   =				2048
  elif "huge" in model_name:
    a_   :						Any   =				1280
    a_   :						List[str]   =				5120
    a_   :						Dict   =				32
    a_   :						int   =				16
    a_   :						int   =				12
    a_   :						Tuple   =				8
    a_   :						str   =				640
    a_   :						List[Any]   =				2560
  elif "base" not in model_name:
    raise ValueError("""Model name should include either \"small\", \"base\", \"large\", or \"huge\"""" )






def 						_snake_case    (   A_							:     Any ):
  """simple docstring"""







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

  return name






def 						_snake_case    (   A_							:     Optional[Any]  ,		A_							:     Any ):
  """simple docstring"""







  for key in orig_state_dict.copy().keys():
    a_   :						Optional[Any]   =				orig_state_dict.pop(__lowerCamelCase )

    if key.startswith("""encoder.""" ):
      a_   :						Optional[Any]   =				key.replace("""encoder."""  ,		"""""" )

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

  return orig_state_dict






def 						_snake_case    (   ):
  """simple docstring"""







  a_   :						Union[str, Any]   =				hf_hub_download(
      repo_id="""hf-internal-testing/spaghetti-video"""  ,		filename="""eating_spaghetti.npy"""  ,		repo_type="""dataset""" )
  a_   :						Optional[Any]   =				np.load(__lowerCamelCase )
  return list(__lowerCamelCase )






def 						_snake_case    (   A_							:     Optional[Any]  ,		A_							:     Union[str, Any]  ,		A_							:     Optional[Any]  ,		A_							:     Any ):
  """simple docstring"""







  a_   :						List[str]   =				get_videomae_config(__lowerCamelCase )

  if "finetuned" in model_name:
    a_   :						List[Any]   =				VideoMAEForVideoClassification(__lowerCamelCase )
  else:
    a_   :						List[str]   =				VideoMAEForPreTraining(__lowerCamelCase )

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

  model.load_state_dict(__lowerCamelCase )
  model.eval()

  # verify model on basic input
  a_   :						Tuple   =				VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]  ,		image_std=[0.5, 0.5, 0.5] )
  a_   :						Dict   =				prepare_video()
  a_   :						Dict   =				image_processor(__lowerCamelCase  ,		return_tensors="""pt""" )

  if "finetuned" not in model_name:
    a_   :						List[Any]   =				hf_hub_download(repo_id="""hf-internal-testing/bool-masked-pos"""  ,		filename="""bool_masked_pos.pt""" )
    a_   :						Union[str, Any]   =				torch.load(__lowerCamelCase )

  a_   :						Any   =				model(**__lowerCamelCase )
  a_   :						List[Any]   =				outputs.logits

  a_   :						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":
    a_   :						List[str]   =				torch.Size([1, 400] )
    a_   :						List[str]   =				torch.tensor([-0.9291, -0.4061, -0.9307] )
  elif model_name == "videomae-small-finetuned-ssv2":
    a_   :						Dict   =				torch.Size([1, 174] )
    a_   :						Tuple   =				torch.tensor([0.2671, -0.4689, -0.8235] )
  elif model_name == "videomae-base":
    a_   :						Union[str, Any]   =				torch.Size([1, 1408, 1536] )
    a_   :						Union[str, 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":
    a_   :						str   =				torch.Size([1, 1408, 1536] )
    a_   :						List[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
    a_   :						int   =				torch.tensor([0.5142] ) if config.norm_pix_loss else torch.tensor([0.6469] )
  elif model_name == "videomae-large":
    a_   :						Optional[int]   =				torch.Size([1, 1408, 1536] )
    a_   :						Tuple   =				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":
    a_   :						Dict   =				torch.Size([1, 400] )
    a_   :						Any   =				torch.tensor([0.0771, 0.0011, -0.3625] )
  elif model_name == "videomae-huge-finetuned-kinetics":
    a_   :						str   =				torch.Size([1, 400] )
    a_   :						List[str]   =				torch.tensor([0.2433, 0.1632, -0.4894] )
  elif model_name == "videomae-base-short-finetuned-kinetics":
    a_   :						int   =				torch.Size([1, 400] )
    a_   :						Tuple   =				torch.tensor([0.6588, 0.0990, -0.2493] )
  elif model_name == "videomae-base-finetuned-kinetics":
    a_   :						Optional[Any]   =				torch.Size([1, 400] )
    a_   :						Dict   =				torch.tensor([0.3669, -0.0688, -0.2421] )
  elif model_name == "videomae-base-short-ssv2":
    a_   :						List[Any]   =				torch.Size([1, 1408, 1536] )
    a_   :						Union[str, 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":
    a_   :						Any   =				torch.Size([1, 174] )
    a_   :						str   =				torch.tensor([-0.0537, -0.1539, -0.3266] )
  elif model_name == "videomae-base-ssv2":
    a_   :						Dict   =				torch.Size([1, 1408, 1536] )
    a_   :						List[Any]   =				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":
    a_   :						List[Any]   =				torch.Size([1, 174] )
    a_   :						Optional[int]   =				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]  ,		__lowerCamelCase  ,		atol=1E-4 )
  else:
    print("""Logits:"""  ,		logits[0, :3, :3] )
    assert torch.allclose(logits[0, :3, :3]  ,		__lowerCamelCase  ,		atol=1E-4 )
  print("""Logits ok!""" )

  # verify loss, if applicable
  if model_name == "videomae-base-short":
    a_   :						List[Any]   =				outputs.loss
    assert torch.allclose(__lowerCamelCase  ,		__lowerCamelCase  ,		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(__lowerCamelCase )
    model.save_pretrained(__lowerCamelCase )

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


if __name__ == "__main__":
    __snake_case:	Union[str, Any]										=      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."
    )

    __snake_case:	Any										=      parser.parse_args()
    convert_videomae_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path, args.model_name, args.push_to_hub)




from __future__ import annotations

from collections.abc import Callable
from typing import Any, Generic, TypeVar

__lowerCamelCase				=			TypeVar("""T""")






class     UpperCAmelCase						(		Generic[T]					):



   def __init__(self						:   int       ,		snake_case__						:   list[T]       ,		snake_case__						:   Callable[[T, T], T]					)						->      None:


          '''simple docstring'''


          snake_case			:  Any | T     							= None

          snake_case			:  int     							= len(snake_case__					)
          snake_case			:  list[T]     							= [any_type for _ in range(self.N					)] + arr
          snake_case			:  str     							= fnc
          self.build()



   def 							_SCREAMING_SNAKE_CASE  (self						:   Union[str, Any]					)						->      None:


          '''simple docstring'''


          for p in range(self.N - 1       ,		0       ,		-1					):
                 snake_case			:  int     							= self.fn(self.st[p * 2]       ,		self.st[p * 2 + 1]					)



   def 							_SCREAMING_SNAKE_CASE  (self						:   Dict       ,		snake_case__						:   int       ,		snake_case__						:   T					)						->      None:


          '''simple docstring'''


          p += self.N
          snake_case			:  int     							= v
          while p > 1:
                 snake_case			:  List[str]     							= p // 2
                 snake_case			:  Tuple     							= self.fn(self.st[p * 2]       ,		self.st[p * 2 + 1]					)



   def 							_SCREAMING_SNAKE_CASE  (self						:   Optional[Any]       ,		snake_case__						:   int       ,		snake_case__						:   int					)						->      T | None:  # noqa: E741


          '''simple docstring'''


          snake_case							,   snake_case			:  Optional[int]     							= l + self.N, r + self.N

          snake_case			:  T | None     							= None
          while l <= r:
                 if l % 2 == 1:
                        snake_case			:  List[str]     							= self.st[l] if res is None else self.fn(snake_case__       ,		self.st[l]					)
                 if r % 2 == 0:
                        snake_case			:  int     							= self.st[r] if res is None else self.fn(snake_case__       ,		self.st[r]					)
                 snake_case							,   snake_case			:  str     							= (l + 1) // 2, (r - 1) // 2
          return res


if __name__ == "__main__":
   from functools import reduce

   __lowerCamelCase				=			[1, 10, -2, 9, -3, 8, 4, -7, 5, 6, 11, -12]

   __lowerCamelCase				=			{
       0: 7,
       1: 2,
       2: 6,
       3: -14,
       4: 5,
       5: 4,
       6: 7,
       7: -10,
       8: 9,
       9: 10,
       10: 12,
       11: 1,
   }

   __lowerCamelCase				=			SegmentTree(test_array, min)
   __lowerCamelCase				=			SegmentTree(test_array, max)
   __lowerCamelCase				=			SegmentTree(test_array, lambda a, b: a + b)


   def      UpperCamelCase   (       ):
          for i in range(len(__lowerCamelCase       )       ):
                 for j in range(__lowerCamelCase							,   len(__lowerCamelCase       )       ):
                        snake_case			:  int     							= reduce(__lowerCamelCase							,   test_array[i : j + 1]       )
                        snake_case			:  Tuple     							= reduce(__lowerCamelCase							,   test_array[i : j + 1]       )
                        snake_case			:  Union[str, Any]     							= reduce(lambda __lowerCamelCase							,   __lowerCamelCase       : a + b							,   test_array[i : j + 1]       )
                        assert min_range == min_segment_tree.query(__lowerCamelCase							,   __lowerCamelCase       )
                        assert max_range == max_segment_tree.query(__lowerCamelCase							,   __lowerCamelCase       )
                        assert sum_range == sum_segment_tree.query(__lowerCamelCase							,   __lowerCamelCase       )

   test_all_segments()

   for index, value in test_updates.items():
      __lowerCamelCase				=			value
      min_segment_tree.update(index, value)
      max_segment_tree.update(index, value)
      sum_segment_tree.update(index, value)
      test_all_segments()





