


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__	:



    """simple docstring"""






    def __init__(      self ,       A_ ,       A_=13 ,       A_=10 ,       A_=3 ,       A_=2 ,       A_=2 ,       A_=2 ,       A_=True ,       A_=True ,       A_=32 ,       A_=5 ,       A_=4 ,       A_=37 ,       A_="gelu" ,       A_=0.1 ,       A_=0.1 ,       A_=10 ,       A_=0.02 ,       A_=0.9 ,       A_=None ,       )  ->	int:
         __UpperCamelCase   							=parent
         __UpperCamelCase   							=batch_size
         __UpperCamelCase   							=image_size
         __UpperCamelCase   							=num_channels
         __UpperCamelCase   							=patch_size
         __UpperCamelCase   							=tubelet_size
         __UpperCamelCase   							=num_frames
         __UpperCamelCase   							=is_training
         __UpperCamelCase   							=use_labels
         __UpperCamelCase   							=hidden_size
         __UpperCamelCase   							=num_hidden_layers
         __UpperCamelCase   							=num_attention_heads
         __UpperCamelCase   							=intermediate_size
         __UpperCamelCase   							=hidden_act
         __UpperCamelCase   							=hidden_dropout_prob
         __UpperCamelCase   							=attention_probs_dropout_prob
         __UpperCamelCase   							=type_sequence_label_size
         __UpperCamelCase   							=initializer_range
         __UpperCamelCase   							=mask_ratio
         __UpperCamelCase   							=scope

         # in VideoMAE, the number of tokens equals num_frames/tubelet_size * num_patches per frame
         __UpperCamelCase   							=(image_size // patch_size) ** 2
         __UpperCamelCase   							=(num_frames // tubelet_size) * self.num_patches_per_frame

         # use this variable to define bool_masked_pos
         __UpperCamelCase   							=int(mask_ratio * self.seq_length   )






    def _a (      self   )  ->	Dict:
         __UpperCamelCase   							=floats_tensor(
             [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]   )

         __UpperCamelCase   							=None
         if self.use_labels:
              __UpperCamelCase   							=ids_tensor([self.batch_size] ,       self.type_sequence_label_size   )

         __UpperCamelCase   							=self.get_config()

         return config, pixel_values, labels






    def _a (      self   )  ->	Optional[Any]:
         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=__A ,       initializer_range=self.initializer_range ,       )






    def _a (      self ,       A_ ,       A_ ,       A_   )  ->	Dict:
         __UpperCamelCase   							=VideoMAEModel(config=__A   )
         model.to(__A   )
         model.eval()
         __UpperCamelCase   							=model(__A   )
         self.parent.assertEqual(result.last_hidden_state.shape ,       (self.batch_size, self.seq_length, self.hidden_size)   )






    def _a (      self ,       A_ ,       A_ ,       A_   )  ->	Optional[int]:
         __UpperCamelCase   							=VideoMAEForPreTraining(__A   )
         model.to(__A   )
         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
         __UpperCamelCase   							=torch.ones((self.num_masks,)   )
         __UpperCamelCase   							=torch.cat([mask, torch.zeros(self.seq_length - mask.size(0   )   )]   )
         __UpperCamelCase   							=mask.expand(self.batch_size ,       -1   ).bool()

         __UpperCamelCase   							=model(__A ,       __A   )
         # model only returns predictions for masked patches
         __UpperCamelCase   							=mask.sum().item()
         __UpperCamelCase   							=3 * self.tubelet_size * self.patch_size**2
         self.parent.assertEqual(result.logits.shape ,       (self.batch_size, num_masked_patches, decoder_num_labels)   )






    def _a (      self   )  ->	Optional[int]:
         __UpperCamelCase   							=self.prepare_config_and_inputs()
         __UpperCamelCase   							=config_and_inputs
         __UpperCamelCase   							={'''pixel_values''': pixel_values}
         return config, inputs_dict






@require_torch
class    UpperCAmelCase__	( lowerCamelCase__       ,  lowerCamelCase__       ,  unittest.TestCase		):



    """simple docstring"""






    UpperCAmelCase__						:		str              =  (
        (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
    )
    UpperCAmelCase__						:		int              =  (
        {"feature-extraction": VideoMAEModel, "video-classification": VideoMAEForVideoClassification}
        if is_torch_available()
        else {}
    )

    UpperCAmelCase__						:		str              =  False
    UpperCAmelCase__						:		Dict              =  False
    UpperCAmelCase__						:		Union[str, Any]              =  False
    UpperCAmelCase__						:		int              =  False






    def _a (      self   )  ->	List[str]:
         __UpperCamelCase   							=VideoMAEModelTester(self   )
         __UpperCamelCase   							=ConfigTester(self ,       config_class=__A ,       has_text_modality=__A ,       hidden_size=37   )






    def _a (      self ,       A_ ,       A_ ,       A_=False   )  ->	List[Any]:
         __UpperCamelCase   							=copy.deepcopy(__A   )

         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
              __UpperCamelCase   							=torch.ones((self.model_tester.num_masks,)   )
              __UpperCamelCase   							=torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0   )   )]   )
              __UpperCamelCase   							=mask.expand(self.model_tester.batch_size ,       -1   ).bool()
              __UpperCamelCase   							=bool_masked_pos.to(__A   )

         if return_labels:
              if model_class in [
                  *get_values(__A   ),
              ]:
                   __UpperCamelCase   							=torch.zeros(
                       self.model_tester.batch_size ,       dtype=torch.long ,       device=__A   )

         return inputs_dict






    def _a (      self   )  ->	List[str]:
         self.config_tester.run_common_tests()






    @unittest.skip(reason='VideoMAE does not use inputs_embeds'   )
    def _a (      self   )  ->	Optional[int]:
         pass






    def _a (      self   )  ->	Optional[int]:
         __UpperCamelCase   							=self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
              __UpperCamelCase   							=model_class(__A   )
              self.assertIsInstance(model.get_input_embeddings() ,       (nn.Module)   )
              __UpperCamelCase   							=model.get_output_embeddings()
              self.assertTrue(x is None or isinstance(__A ,       nn.Linear   )   )






    def _a (      self   )  ->	str:
         __UpperCamelCase   							=self.model_tester.prepare_config_and_inputs_for_common()

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

              __UpperCamelCase   							=['''pixel_values''']
              self.assertListEqual(arg_names[:1] ,       __A   )






    def _a (      self   )  ->	Dict:
         __UpperCamelCase   							=self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*__A   )






    def _a (      self   )  ->	List[str]:
         __UpperCamelCase   							=self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_pretraining(*__A   )






    @slow
    def _a (      self   )  ->	Union[str, Any]:
         for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
              __UpperCamelCase   							=VideoMAEModel.from_pretrained(__A   )
              self.assertIsNotNone(__A   )






    def _a (      self   )  ->	str:
         if not self.has_attentions:
              pass

         else:
              __UpperCamelCase   							=self.model_tester.prepare_config_and_inputs_for_common()
              __UpperCamelCase   							=True

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

                   __UpperCamelCase   							=True
                   __UpperCamelCase   							=False
                   __UpperCamelCase   							=True
                   __UpperCamelCase   							=model_class(__A   )
                   model.to(__A   )
                   model.eval()
                   with torch.no_grad():
                        __UpperCamelCase   							=model(**self._prepare_for_class(__A ,       __A   )   )
                   __UpperCamelCase   							=outputs.attentions
                   self.assertEqual(len(__A   ) ,       self.model_tester.num_hidden_layers   )

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

                   self.assertListEqual(
                       list(attentions[0].shape[-3:]   ) ,       [self.model_tester.num_attention_heads, seq_len, seq_len] ,       )
                   __UpperCamelCase   							=len(__A   )

                   # Check attention is always last and order is fine
                   __UpperCamelCase   							=True
                   __UpperCamelCase   							=True
                   __UpperCamelCase   							=model_class(__A   )
                   model.to(__A   )
                   model.eval()
                   with torch.no_grad():
                        __UpperCamelCase   							=model(**self._prepare_for_class(__A ,       __A   )   )

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

                   __UpperCamelCase   							=outputs.attentions

                   self.assertEqual(len(__A   ) ,       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 _a (      self   )  ->	Optional[Any]:
         def check_hidden_states_output(A_ ,       A_ ,       A_   ):
              __UpperCamelCase   							=model_class(__A   )
              model.to(__A   )
              model.eval()

              with torch.no_grad():
                   __UpperCamelCase   							=model(**self._prepare_for_class(__A ,       __A   )   )

              __UpperCamelCase   							=outputs.hidden_states
              __UpperCamelCase   							=self.model_tester.num_hidden_layers + 1
              self.assertEqual(len(__A   ) ,       __A   )

              __UpperCamelCase   							=self.model_tester.seq_length - self.model_tester.num_masks
              __UpperCamelCase   							=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] ,       )

         __UpperCamelCase   							=self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
              __UpperCamelCase   							=True
              check_hidden_states_output(__A ,       __A ,       __A   )

              # check that output_hidden_states also work using config
              del inputs_dict["output_hidden_states"]
              __UpperCamelCase   							=True

              check_hidden_states_output(__A ,       __A ,       __A   )






    @unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.'   )
    def _a (      self   )  ->	Optional[int]:
         pass

def 						_UpperCAmelCase    (      ):
     __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__    )






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



    """simple docstring"""






    @cached_property
    def _a (      self   )  ->	Dict:
         # logits were tested with a different mean and std, so we use the same here
         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 _a (      self   )  ->	Optional[Any]:
         __UpperCamelCase   							=VideoMAEForVideoClassification.from_pretrained('MCG-NJU/videomae-base-finetuned-kinetics'   ).to(
             __A   )

         __UpperCamelCase   							=self.default_image_processor
         __UpperCamelCase   							=prepare_video()
         __UpperCamelCase   							=image_processor(__A ,       return_tensors='pt'   ).to(__A   )

         # forward pass
         with torch.no_grad():
              __UpperCamelCase   							=model(**__A   )

         # verify the logits
         __UpperCamelCase   							=torch.Size((1, 400)   )
         self.assertEqual(outputs.logits.shape ,       __A   )

         __UpperCamelCase   							=torch.tensor([0.3669, -0.0688, -0.2421]   ).to(__A   )

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






    @slow
    def _a (      self   )  ->	List[str]:
         __UpperCamelCase   							=VideoMAEForPreTraining.from_pretrained('MCG-NJU/videomae-base-short'   ).to(__A   )

         __UpperCamelCase   							=self.default_image_processor
         __UpperCamelCase   							=prepare_video()
         __UpperCamelCase   							=image_processor(__A ,       return_tensors='pt'   ).to(__A   )

         # add boolean mask, indicating which patches to mask
         __UpperCamelCase   							=hf_hub_download(repo_id='hf-internal-testing/bool-masked-pos' ,       filename='bool_masked_pos.pt'   )
         __UpperCamelCase   							=torch.load(__A   )

         # forward pass
         with torch.no_grad():
              __UpperCamelCase   							=model(**__A   )

         # verify the logits
         __UpperCamelCase   							=torch.Size([1, 1408, 1536]   )
         __UpperCamelCase   							=torch.tensor(
             [[0.7994, 0.9612, 0.8508], [0.7401, 0.8958, 0.8302], [0.5862, 0.7468, 0.7325]] ,       device=__A   )
         self.assertEqual(outputs.logits.shape ,       __A   )
         self.assertTrue(torch.allclose(outputs.logits[0, :3, :3] ,       __A ,       atol=1E-4   )   )

         # verify the loss (`config.norm_pix_loss` = `True`)
         __UpperCamelCase   							=torch.tensor([0.5142] ,       device=__A   )
         self.assertTrue(torch.allclose(outputs.loss ,       __A ,       atol=1E-4   )   )

         # verify the loss (`config.norm_pix_loss` = `False`)
         __UpperCamelCase   							=VideoMAEForPreTraining.from_pretrained('MCG-NJU/videomae-base-short' ,       norm_pix_loss=__A   ).to(
             __A   )

         with torch.no_grad():
              __UpperCamelCase   							=model(**__A   )

         __UpperCamelCase   							=torch.tensor(torch.tensor([0.6469]   ) ,       device=__A   )
         self.assertTrue(torch.allclose(outputs.loss ,       __A ,       atol=1E-4   )   )









# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse

from ...utils.dataclasses import (
    ComputeEnvironment,
    DistributedType,
    DynamoBackend,
    PrecisionType,
    SageMakerDistributedType,
)
from ..menu import BulletMenu


_A			=						[
    'EAGER',
    'AOT_EAGER',
    'INDUCTOR',
    'NVFUSER',
    'AOT_NVFUSER',
    'AOT_CUDAGRAPHS',
    'OFI',
    'FX2TRT',
    'ONNXRT',
    'IPEX',
]

def 						_UpperCAmelCase    (      SCREAMING_SNAKE_CASE__      : Optional[int]      ,					SCREAMING_SNAKE_CASE__      : Optional[Any]=None      ,					SCREAMING_SNAKE_CASE__      : Optional[Any]=None      ,					SCREAMING_SNAKE_CASE__      : str=None    ):
     __UpperCamelCase   							=True
     while ask_again:
          __UpperCamelCase   							=input(SCREAMING_SNAKE_CASE__    )
          try:
               if default is not None and len(SCREAMING_SNAKE_CASE__    ) == 0:
                    return default
               return convert_value(SCREAMING_SNAKE_CASE__    ) if convert_value is not None else result
          except Exception:
               if error_message is not None:
                    print(SCREAMING_SNAKE_CASE__    )

def 						_UpperCAmelCase    (      SCREAMING_SNAKE_CASE__      : List[Any]      ,					SCREAMING_SNAKE_CASE__      : List[str]=[]      ,					SCREAMING_SNAKE_CASE__      : Union[str, Any]=None      ,					SCREAMING_SNAKE_CASE__      : str=0    ):
     __UpperCamelCase   							=BulletMenu(SCREAMING_SNAKE_CASE__      ,					SCREAMING_SNAKE_CASE__    )
     __UpperCamelCase   							=menu.run(default_choice=SCREAMING_SNAKE_CASE__    )
     return convert_value(SCREAMING_SNAKE_CASE__    ) if convert_value is not None else result

def 						_UpperCAmelCase    (      SCREAMING_SNAKE_CASE__      : Union[str, Any]    ):
     __UpperCamelCase   							=int(SCREAMING_SNAKE_CASE__    )
     return ComputeEnvironment(['LOCAL_MACHINE', 'AMAZON_SAGEMAKER'][value]    )

def 						_UpperCAmelCase    (      SCREAMING_SNAKE_CASE__      : List[str]    ):
     __UpperCamelCase   							=int(SCREAMING_SNAKE_CASE__    )
     return DistributedType(['NO', 'MULTI_CPU', 'MULTI_XPU', 'MULTI_GPU', 'MULTI_NPU', 'TPU'][value]    )

def 						_UpperCAmelCase    (      SCREAMING_SNAKE_CASE__      : Optional[int]    ):
     __UpperCamelCase   							=int(SCREAMING_SNAKE_CASE__    )
     return DynamoBackend(DYNAMO_BACKENDS[value]    ).value

def 						_UpperCAmelCase    (      SCREAMING_SNAKE_CASE__      : Dict    ):
     __UpperCamelCase   							=int(SCREAMING_SNAKE_CASE__    )
     return PrecisionType(['no', 'fp16', 'bf16', 'fp8'][value]    )

def 						_UpperCAmelCase    (      SCREAMING_SNAKE_CASE__      : Optional[Any]    ):
     __UpperCamelCase   							=int(SCREAMING_SNAKE_CASE__    )
     return SageMakerDistributedType(['NO', 'DATA_PARALLEL', 'MODEL_PARALLEL'][value]    )

def 						_UpperCAmelCase    (      SCREAMING_SNAKE_CASE__      : Any    ):
     return {"yes": True, "no": False}[value.lower()]






class    UpperCAmelCase__	( argparse.RawDescriptionHelpFormatter		):



    """simple docstring"""






    def _a (      self ,       A_ ,       A_ ,       A_ ,       A_   )  ->	Optional[int]:
         __UpperCamelCase   							=super()._format_usage(A_ ,       A_ ,       A_ ,       A_   )
         __UpperCamelCase   							=usage.replace('<command> [<args>] ' ,       ''   )
         return usage





