




'''simple docstring'''
import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import VideoMAEConfig
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
  import torch
  from torch import nn

  from transformers import (
      MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING,
      VideoMAEForPreTraining,
      VideoMAEForVideoClassification,
      VideoMAEModel,
  )
  from transformers.models.videomae.modeling_videomae import VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
  from transformers import VideoMAEImageProcessor






class    UpperCAmelCase_				:





 def __init__(    self ,       lowercase_ ,       lowercase_=13 ,       lowercase_=10 ,       lowercase_=3 ,       lowercase_=2 ,       lowercase_=2 ,       lowercase_=2 ,       lowercase_=True ,       lowercase_=True ,       lowercase_=32 ,       lowercase_=5 ,       lowercase_=4 ,       lowercase_=37 ,       lowercase_="gelu" ,       lowercase_=0.1 ,       lowercase_=0.1 ,       lowercase_=10 ,       lowercase_=0.02 ,       lowercase_=0.9 ,       lowercase_=None ,       ):
     snake_case_ :    Union[str, Any]     			=						parent
     snake_case_ :    Union[str, Any]     			=						batch_size
     snake_case_ :    int     			=						image_size
     snake_case_ :    Optional[int]     			=						num_channels
     snake_case_ :    int     			=						patch_size
     snake_case_ :    Any     			=						tubelet_size
     snake_case_ :    Tuple     			=						num_frames
     snake_case_ :    Union[str, Any]     			=						is_training
     snake_case_ :    Optional[Any]     			=						use_labels
     snake_case_ :    Tuple     			=						hidden_size
     snake_case_ :    Union[str, Any]     			=						num_hidden_layers
     snake_case_ :    Union[str, Any]     			=						num_attention_heads
     snake_case_ :    Optional[int]     			=						intermediate_size
     snake_case_ :    List[str]     			=						hidden_act
     snake_case_ :    Any     			=						hidden_dropout_prob
     snake_case_ :    Union[str, Any]     			=						attention_probs_dropout_prob
     snake_case_ :    Dict     			=						type_sequence_label_size
     snake_case_ :    Tuple     			=						initializer_range
     snake_case_ :    Tuple     			=						mask_ratio
     snake_case_ :    List[str]     			=						scope

     # in VideoMAE, the number of tokens equals num_frames/tubelet_size * num_patches per frame
     snake_case_ :    Optional[Any]     			=						(image_size // patch_size) ** 2
     snake_case_ :    str     			=						(num_frames // tubelet_size) * self.num_patches_per_frame

     # use this variable to define bool_masked_pos
     snake_case_ :    List[str]     			=						int(mask_ratio * self.seq_length)





 def        snake_case__		(    self):
     snake_case_ :    List[Any]     			=						floats_tensor(
         [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size])

     snake_case_ :    List[Any]     			=						None
     if self.use_labels:
         snake_case_ :    Tuple     			=						ids_tensor([self.batch_size] ,       self.type_sequence_label_size)

     snake_case_ :    Dict     			=						self.get_config()

     return config, pixel_values, labels





 def        snake_case__		(    self):
     return VideoMAEConfig(
         image_size=self.image_size ,       patch_size=self.patch_size ,       num_channels=self.num_channels ,       num_frames=self.num_frames ,       tubelet_size=self.tubelet_size ,       hidden_size=self.hidden_size ,       num_hidden_layers=self.num_hidden_layers ,       num_attention_heads=self.num_attention_heads ,       intermediate_size=self.intermediate_size ,       hidden_act=self.hidden_act ,       hidden_dropout_prob=self.hidden_dropout_prob ,       attention_probs_dropout_prob=self.attention_probs_dropout_prob ,       is_decoder=_lowerCAmelCase ,       initializer_range=self.initializer_range ,       )





 def        snake_case__		(    self ,       lowercase_ ,       lowercase_ ,       lowercase_):
     snake_case_ :    Optional[Any]     			=						VideoMAEModel(config=_lowerCAmelCase)
     model.to(_lowerCAmelCase)
     model.eval()
     snake_case_ :    Union[str, Any]     			=						model(_lowerCAmelCase)
     self.parent.assertEqual(result.last_hidden_state.shape ,       (self.batch_size, self.seq_length, self.hidden_size))





 def        snake_case__		(    self ,       lowercase_ ,       lowercase_ ,       lowercase_):
     snake_case_ :    Union[str, Any]     			=						VideoMAEForPreTraining(_lowerCAmelCase)
     model.to(_lowerCAmelCase)
     model.eval()
     # important: each video needs to have the same number of masked patches
     # hence we define a single mask, which we then repeat for each example in the batch
     snake_case_ :    Dict     			=						torch.ones((self.num_masks,))
     snake_case_ :    Optional[int]     			=						torch.cat([mask, torch.zeros(self.seq_length - mask.size(0))])
     snake_case_ :    Any     			=						mask.expand(self.batch_size ,       -1).bool()

     snake_case_ :    List[str]     			=						model(_lowerCAmelCase ,       _lowerCAmelCase)
     # model only returns predictions for masked patches
     snake_case_ :    Union[str, Any]     			=						mask.sum().item()
     snake_case_ :    Tuple     			=						3 * self.tubelet_size * self.patch_size**2
     self.parent.assertEqual(result.logits.shape ,       (self.batch_size, num_masked_patches, decoder_num_labels))





 def        snake_case__		(    self):
     snake_case_ :    str     			=						self.prepare_config_and_inputs()
     snake_case_			, snake_case_			, snake_case_ :    Any     			=						config_and_inputs
     snake_case_ :    Optional[Any]     			=						{"pixel_values": pixel_values}
     return config, inputs_dict






@require_torch
class    UpperCAmelCase_				(	_lowerCAmelCase    ,						_lowerCAmelCase    ,						unittest.TestCase					):
 UpperCAmelCase_		    =						(
     (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
 )
 UpperCAmelCase_		    =						(
     {'feature-extraction': VideoMAEModel, 'video-classification': VideoMAEForVideoClassification}
     if is_torch_available()
     else {}
 )

 UpperCAmelCase_		    =						False
 UpperCAmelCase_		    =						False
 UpperCAmelCase_		    =						False
 UpperCAmelCase_		    =						False





 def        snake_case__		(    self):
     snake_case_ :    str     			=						VideoMAEModelTester(self)
     snake_case_ :    str     			=						ConfigTester(self ,       config_class=_lowerCAmelCase ,       has_text_modality=_lowerCAmelCase ,       hidden_size=37)





 def        snake_case__		(    self ,       lowercase_ ,       lowercase_ ,       lowercase_=False):
     snake_case_ :    List[str]     			=						copy.deepcopy(_lowerCAmelCase)

     if model_class == VideoMAEForPreTraining:
         # important: each video needs to have the same number of masked patches
         # hence we define a single mask, which we then repeat for each example in the batch
         snake_case_ :    List[str]     			=						torch.ones((self.model_tester.num_masks,))
         snake_case_ :    List[str]     			=						torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0))])
         snake_case_ :    Tuple     			=						mask.expand(self.model_tester.batch_size ,       -1).bool()
         snake_case_ :    str     			=						bool_masked_pos.to(_lowerCAmelCase)

     if return_labels:
         if model_class in [
             *get_values(_lowerCAmelCase),
         ]:
             snake_case_ :    Optional[int]     			=						torch.zeros(
                 self.model_tester.batch_size ,       dtype=torch.long ,       device=_lowerCAmelCase)

     return inputs_dict





 def        snake_case__		(    self):
     self.config_tester.run_common_tests()





 @unittest.skip(reason="VideoMAE does not use inputs_embeds")
 def        snake_case__		(    self):
     pass





 def        snake_case__		(    self):
     snake_case_			, snake_case_ :    List[str]     			=						self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
         snake_case_ :    List[Any]     			=						model_class(_lowerCAmelCase)
         self.assertIsInstance(model.get_input_embeddings() ,       (nn.Module))
         snake_case_ :    Dict     			=						model.get_output_embeddings()
         self.assertTrue(x is None or isinstance(_lowerCAmelCase ,       nn.Linear))





 def        snake_case__		(    self):
     snake_case_			, snake_case_ :    int     			=						self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
         snake_case_ :    Any     			=						model_class(_lowerCAmelCase)
         snake_case_ :    Optional[Any]     			=						inspect.signature(model.forward)
         # signature.parameters is an OrderedDict => so arg_names order is deterministic
         snake_case_ :    Optional[int]     			=						[*signature.parameters.keys()]

         snake_case_ :    Tuple     			=						["pixel_values"]
         self.assertListEqual(arg_names[:1] ,       _lowerCAmelCase)





 def        snake_case__		(    self):
     snake_case_ :    Any     			=						self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_model(*_lowerCAmelCase)





 def        snake_case__		(    self):
     snake_case_ :    List[str]     			=						self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_pretraining(*_lowerCAmelCase)





 @slow
 def        snake_case__		(    self):







     for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
         snake_case_ :    Union[str, Any]     			=						VideoMAEModel.from_pretrained(_lowerCAmelCase)
         self.assertIsNotNone(_lowerCAmelCase)





 def        snake_case__		(    self):







     if not self.has_attentions:
         pass

     else:
         snake_case_			, snake_case_ :    Optional[int]     			=						self.model_tester.prepare_config_and_inputs_for_common()
         snake_case_ :    Dict     			=						True

         for model_class in self.all_model_classes:
             snake_case_ :    Tuple     			=						self.model_tester.seq_length - self.model_tester.num_masks
             snake_case_ :    str     			=						(
                 num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length
             )

             snake_case_ :    Dict     			=						True
             snake_case_ :    Any     			=						False
             snake_case_ :    List[str]     			=						True
             snake_case_ :    Union[str, Any]     			=						model_class(_lowerCAmelCase)
             model.to(_lowerCAmelCase)
             model.eval()
             with torch.no_grad():
                 snake_case_ :    Dict     			=						model(**self._prepare_for_class(_lowerCAmelCase ,       _lowerCAmelCase))
             snake_case_ :    str     			=						outputs.attentions
             self.assertEqual(len(_lowerCAmelCase) ,       self.model_tester.num_hidden_layers)

             # check that output_attentions also work using config
             del inputs_dict["output_attentions"]
             snake_case_ :    Dict     			=						True
             snake_case_ :    List[str]     			=						model_class(_lowerCAmelCase)
             model.to(_lowerCAmelCase)
             model.eval()
             with torch.no_grad():
                 snake_case_ :    Any     			=						model(**self._prepare_for_class(_lowerCAmelCase ,       _lowerCAmelCase))
             snake_case_ :    List[Any]     			=						outputs.attentions
             self.assertEqual(len(_lowerCAmelCase) ,       self.model_tester.num_hidden_layers)

             self.assertListEqual(
                 list(attentions[0].shape[-3:]) ,       [self.model_tester.num_attention_heads, seq_len, seq_len] ,       )
             snake_case_ :    Optional[Any]     			=						len(_lowerCAmelCase)

             # Check attention is always last and order is fine
             snake_case_ :    List[Any]     			=						True
             snake_case_ :    str     			=						True
             snake_case_ :    Optional[int]     			=						model_class(_lowerCAmelCase)
             model.to(_lowerCAmelCase)
             model.eval()
             with torch.no_grad():
                 snake_case_ :    Any     			=						model(**self._prepare_for_class(_lowerCAmelCase ,       _lowerCAmelCase))

             self.assertEqual(out_len + 1 ,       len(_lowerCAmelCase))

             snake_case_ :    List[str]     			=						outputs.attentions

             self.assertEqual(len(_lowerCAmelCase) ,       self.model_tester.num_hidden_layers)
             self.assertListEqual(
                 list(self_attentions[0].shape[-3:]) ,       [self.model_tester.num_attention_heads, seq_len, seq_len] ,       )





 def        snake_case__		(    self):







     def check_hidden_states_output(lowercase_ ,       lowercase_ ,       lowercase_):
         snake_case_ :    str     			=						model_class(_lowerCAmelCase)
         model.to(_lowerCAmelCase)
         model.eval()

         with torch.no_grad():
             snake_case_ :    str     			=						model(**self._prepare_for_class(_lowerCAmelCase ,       _lowerCAmelCase))

         snake_case_ :    List[Any]     			=						outputs.hidden_states
         snake_case_ :    Tuple     			=						self.model_tester.num_hidden_layers + 1
         self.assertEqual(len(_lowerCAmelCase) ,       _lowerCAmelCase)

         snake_case_ :    int     			=						self.model_tester.seq_length - self.model_tester.num_masks
         snake_case_ :    List[Any]     			=						num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length

         self.assertListEqual(
             list(hidden_states[0].shape[-2:]) ,       [seq_length, self.model_tester.hidden_size] ,       )

     snake_case_			, snake_case_ :    Union[str, Any]     			=						self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
         snake_case_ :    List[Any]     			=						True
         check_hidden_states_output(_lowerCAmelCase ,       _lowerCAmelCase ,       _lowerCAmelCase)

         # check that output_hidden_states also work using config
         del inputs_dict["output_hidden_states"]
         snake_case_ :    Any     			=						True

         check_hidden_states_output(_lowerCAmelCase ,       _lowerCAmelCase ,       _lowerCAmelCase)





 @unittest.skip("Will be fixed soon by reducing the size of the model used for common tests.")
 def        snake_case__		(    self):
     pass


def 						UpperCamelCase_     (	):
    """simple docstring"""






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






@require_torch
@require_vision
class    UpperCAmelCase_				(	unittest.TestCase					):





 @cached_property
 def        snake_case__		(    self):
     return (
         VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5] ,       image_std=[0.5, 0.5, 0.5])
         if is_vision_available()
         else None
     )





 @slow
 def        snake_case__		(    self):
     snake_case_ :    Tuple     			=						VideoMAEForVideoClassification.from_pretrained("MCG-NJU/videomae-base-finetuned-kinetics").to(
         _lowerCAmelCase)

     snake_case_ :    List[Any]     			=						self.default_image_processor
     snake_case_ :    Dict     			=						prepare_video()
     snake_case_ :    int     			=						image_processor(_lowerCAmelCase ,       return_tensors="pt").to(_lowerCAmelCase)

     # forward pass
     with torch.no_grad():
         snake_case_ :    Optional[int]     			=						model(**_lowerCAmelCase)

     # verify the logits
     snake_case_ :    Tuple     			=						torch.Size((1, 4_00))
     self.assertEqual(outputs.logits.shape ,       _lowerCAmelCase)

     snake_case_ :    Optional[Any]     			=						torch.tensor([0.3_669, -0.0_688, -0.2_421]).to(_lowerCAmelCase)

     self.assertTrue(torch.allclose(outputs.logits[0, :3] ,       _lowerCAmelCase ,       atol=1E-4))





 @slow
 def        snake_case__		(    self):
     snake_case_ :    Union[str, Any]     			=						VideoMAEForPreTraining.from_pretrained("MCG-NJU/videomae-base-short").to(_lowerCAmelCase)

     snake_case_ :    List[str]     			=						self.default_image_processor
     snake_case_ :    List[str]     			=						prepare_video()
     snake_case_ :    Tuple     			=						image_processor(_lowerCAmelCase ,       return_tensors="pt").to(_lowerCAmelCase)

     # add boolean mask, indicating which patches to mask
     snake_case_ :    Optional[Any]     			=						hf_hub_download(repo_id="hf-internal-testing/bool-masked-pos" ,       filename="bool_masked_pos.pt")
     snake_case_ :    str     			=						torch.load(_lowerCAmelCase)

     # forward pass
     with torch.no_grad():
         snake_case_ :    Optional[int]     			=						model(**_lowerCAmelCase)

     # verify the logits
     snake_case_ :    Tuple     			=						torch.Size([1, 14_08, 15_36])
     snake_case_ :    Dict     			=						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]] ,       device=_lowerCAmelCase)
     self.assertEqual(outputs.logits.shape ,       _lowerCAmelCase)
     self.assertTrue(torch.allclose(outputs.logits[0, :3, :3] ,       _lowerCAmelCase ,       atol=1E-4))

     # verify the loss (`config.norm_pix_loss` = `True`)
     snake_case_ :    List[str]     			=						torch.tensor([0.5_142] ,       device=_lowerCAmelCase)
     self.assertTrue(torch.allclose(outputs.loss ,       _lowerCAmelCase ,       atol=1E-4))

     # verify the loss (`config.norm_pix_loss` = `False`)
     snake_case_ :    Any     			=						VideoMAEForPreTraining.from_pretrained("MCG-NJU/videomae-base-short" ,       norm_pix_loss=_lowerCAmelCase).to(
         _lowerCAmelCase)

     with torch.no_grad():
         snake_case_ :    Any     			=						model(**_lowerCAmelCase)

     snake_case_ :    List[str]     			=						torch.tensor(torch.tensor([0.6_469]) ,       device=_lowerCAmelCase)
     self.assertTrue(torch.allclose(outputs.loss ,       _lowerCAmelCase ,       atol=1E-4))






'''simple docstring'''
import argparse
import os

import torch

from transformers.utils import WEIGHTS_NAME


a_											=						["small", "medium", "large"]

a_											=						"lm_head.decoder.weight"
a_											=						"lm_head.weight"

def 						UpperCamelCase_     (	__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE       ):
    """simple docstring"""






    snake_case_ :    List[Any]     			=						torch.load(__SCREAMING_SNAKE_CASE       )
    snake_case_ :    List[Any]     			=						d.pop(__SCREAMING_SNAKE_CASE       )
    os.makedirs(__SCREAMING_SNAKE_CASE,		exist_ok=__SCREAMING_SNAKE_CASE       )
    torch.save(__SCREAMING_SNAKE_CASE,		os.path.join(__SCREAMING_SNAKE_CASE,		__SCREAMING_SNAKE_CASE       )       )


if __name__ == "__main__":
  a_											=						argparse.ArgumentParser()
  parser.add_argument("--dialogpt_path", default=".", type=str)
  a_											=						parser.parse_args()
  for MODEL in DIALOGPT_MODELS:
    a_											=						os.path.join(args.dialogpt_path, f"""{MODEL}_ft.pkl""")
    a_											=						f"""./DialoGPT-{MODEL}"""
    convert_dialogpt_checkpoint(
        checkpoint_path,
        pytorch_dump_folder_path,
    )




