






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

  def __init__(    self						:	int   ,				lowerCamelCase__						:	Union[str, Any]   ,				lowerCamelCase__						:	Union[str, Any]=13   ,				lowerCamelCase__						:	List[Any]=10   ,				lowerCamelCase__						:	Optional[Any]=3   ,				lowerCamelCase__						:	List[str]=2   ,				lowerCamelCase__						:	Optional[Any]=2   ,				lowerCamelCase__						:	str=2   ,				lowerCamelCase__						:	List[str]=True   ,				lowerCamelCase__						:	str=True   ,				lowerCamelCase__						:	Optional[Any]=32   ,				lowerCamelCase__						:	List[Any]=5   ,				lowerCamelCase__						:	Dict=4   ,				lowerCamelCase__						:	str=37   ,				lowerCamelCase__						:	Dict="gelu"   ,				lowerCamelCase__						:	Union[str, Any]=0.1   ,				lowerCamelCase__						:	int=0.1   ,				lowerCamelCase__						:	Tuple=10   ,				lowerCamelCase__						:	List[str]=0.02   ,				lowerCamelCase__						:	Dict=0.9   ,				lowerCamelCase__						:	List[Any]=None   ,				):
     """simple docstring"""

     __UpperCamelCase	:			Union[str, Any]					=					parent
     __UpperCamelCase	:			Tuple					=					batch_size
     __UpperCamelCase	:			Union[str, Any]					=					image_size
     __UpperCamelCase	:			Any					=					num_channels
     __UpperCamelCase	:			List[Any]					=					patch_size
     __UpperCamelCase	:			List[str]					=					tubelet_size
     __UpperCamelCase	:			List[Any]					=					num_frames
     __UpperCamelCase	:			str					=					is_training
     __UpperCamelCase	:			int					=					use_labels
     __UpperCamelCase	:			Any					=					hidden_size
     __UpperCamelCase	:			Any					=					num_hidden_layers
     __UpperCamelCase	:			Tuple					=					num_attention_heads
     __UpperCamelCase	:			Optional[Any]					=					intermediate_size
     __UpperCamelCase	:			Any					=					hidden_act
     __UpperCamelCase	:			Union[str, Any]					=					hidden_dropout_prob
     __UpperCamelCase	:			Dict					=					attention_probs_dropout_prob
     __UpperCamelCase	:			str					=					type_sequence_label_size
     __UpperCamelCase	:			List[str]					=					initializer_range
     __UpperCamelCase	:			Tuple					=					mask_ratio
     __UpperCamelCase	:			Optional[int]					=					scope

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

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

  def    a					(    self						:	str			):
     """simple docstring"""

     __UpperCamelCase	:			Optional[int]					=					floats_tensor(
         [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]			)

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

     __UpperCamelCase	:			Optional[Any]					=					self.get_config()

     return config, pixel_values, labels

  def    a					(    self						:	Tuple			):
     """simple docstring"""

     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    a					(    self						:	int   ,				lowerCamelCase__						:	int   ,				lowerCamelCase__						:	Union[str, Any]   ,				lowerCamelCase__						:	Any			):
     """simple docstring"""

     __UpperCamelCase	:			Optional[Any]					=					VideoMAEModel(config=lowerCamelCase__			)
     model.to(lowerCamelCase__			)
     model.eval()
     __UpperCamelCase	:			str					=					model(lowerCamelCase__			)
     self.parent.assertEqual(result.last_hidden_state.shape   ,				(self.batch_size, self.seq_length, self.hidden_size)			)

  def    a					(    self						:	Optional[int]   ,				lowerCamelCase__						:	Union[str, Any]   ,				lowerCamelCase__						:	Any   ,				lowerCamelCase__						:	Dict			):
     """simple docstring"""

     __UpperCamelCase	:			Tuple					=					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
     __UpperCamelCase	:			int					=					torch.ones((self.num_masks,)			)
     __UpperCamelCase	:			Optional[Any]					=					torch.cat([mask, torch.zeros(self.seq_length - mask.size(0			)			)]			)
     __UpperCamelCase	:			Optional[int]					=					mask.expand(self.batch_size   ,				-1			).bool()

     __UpperCamelCase	:			int					=					model(lowerCamelCase__   ,				lowerCamelCase__			)
     # model only returns predictions for masked patches
     __UpperCamelCase	:			Dict					=					mask.sum().item()
     __UpperCamelCase	:			int					=					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						:	Tuple			):
     """simple docstring"""

     __UpperCamelCase	:			List[str]					=					self.prepare_config_and_inputs()
     __UpperCamelCase	,     __UpperCamelCase	,     __UpperCamelCase	:			Any					=					config_and_inputs
     __UpperCamelCase	:			Union[str, Any]					=					{"""pixel_values""": pixel_values}
     return config, inputs_dict







@require_torch
class 		_A						( UpperCAmelCase_	,				UpperCAmelCase_	,				unittest.TestCase ):
  lowercase_						:			str						=					(
      (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
  )
  lowercase_						:			Union[str, Any]						=					(
      {'''feature-extraction''': VideoMAEModel, '''video-classification''': VideoMAEForVideoClassification}
      if is_torch_available()
      else {}
  )

  lowercase_						:			Union[str, Any]						=					False
  lowercase_						:			Optional[Any]						=					False
  lowercase_						:			int						=					False
  lowercase_						:			List[Any]						=					False

  def    a					(    self						:	List[str]			):
     """simple docstring"""

     __UpperCamelCase	:			List[str]					=					VideoMAEModelTester(self			)
     __UpperCamelCase	:			Dict					=					ConfigTester(self   ,				config_class=lowerCamelCase__   ,				has_text_modality=lowerCamelCase__   ,				hidden_size=37			)

  def    a					(    self						:	Tuple   ,				lowerCamelCase__						:	List[str]   ,				lowerCamelCase__						:	Tuple   ,				lowerCamelCase__						:	Optional[int]=False			):
     """simple docstring"""

     __UpperCamelCase	:			List[Any]					=					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
        __UpperCamelCase	:			Tuple					=					torch.ones((self.model_tester.num_masks,)			)
        __UpperCamelCase	:			str					=					torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0			)			)]			)
        __UpperCamelCase	:			Dict					=					mask.expand(self.model_tester.batch_size   ,				-1			).bool()
        __UpperCamelCase	:			int					=					bool_masked_pos.to(lowerCamelCase__			)

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

     return inputs_dict

  def    a					(    self						:	List[Any]			):
     """simple docstring"""

     self.config_tester.run_common_tests()

  @unittest.skip(reason="""VideoMAE does not use inputs_embeds"""			)
  def    a					(    self						:	List[Any]			):
     """simple docstring"""

     pass

  def    a					(    self						:	Union[str, Any]			):
     """simple docstring"""

     __UpperCamelCase	,     __UpperCamelCase	:			Dict					=					self.model_tester.prepare_config_and_inputs_for_common()

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

  def    a					(    self						:	Tuple			):
     """simple docstring"""

     __UpperCamelCase	,     __UpperCamelCase	:			Dict					=					self.model_tester.prepare_config_and_inputs_for_common()

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

        __UpperCamelCase	:			List[Any]					=					["""pixel_values"""]
        self.assertListEqual(arg_names[:1]   ,				lowerCamelCase__			)

  def    a					(    self						:	Any			):
     """simple docstring"""

     __UpperCamelCase	:			Any					=					self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_model(*lowerCamelCase__			)

  def    a					(    self						:	List[Any]			):
     """simple docstring"""

     __UpperCamelCase	:			List[str]					=					self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_pretraining(*lowerCamelCase__			)

  @slow
  def    a					(    self						:	Dict			):
     """simple docstring"""

     for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        __UpperCamelCase	:			Optional[int]					=					VideoMAEModel.from_pretrained(lowerCamelCase__			)
        self.assertIsNotNone(lowerCamelCase__			)

  def    a					(    self						:	int			):
     """simple docstring"""

     if not self.has_attentions:
        pass

     else:
        __UpperCamelCase	,     __UpperCamelCase	:			Dict					=					self.model_tester.prepare_config_and_inputs_for_common()
        __UpperCamelCase	:			Optional[Any]					=					True

        for model_class in self.all_model_classes:
           __UpperCamelCase	:			Dict					=					self.model_tester.seq_length - self.model_tester.num_masks
           __UpperCamelCase	:			Optional[int]					=					(
               num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length
           )

           __UpperCamelCase	:			str					=					True
           __UpperCamelCase	:			Any					=					False
           __UpperCamelCase	:			Optional[Any]					=					True
           __UpperCamelCase	:			List[str]					=					model_class(lowerCamelCase__			)
           model.to(lowerCamelCase__			)
           model.eval()
           with torch.no_grad():
              __UpperCamelCase	:			Dict					=					model(**self._prepare_for_class(lowerCamelCase__   ,				lowerCamelCase__			)			)
           __UpperCamelCase	:			Any					=					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"]
           __UpperCamelCase	:			Union[str, Any]					=					True
           __UpperCamelCase	:			List[Any]					=					model_class(lowerCamelCase__			)
           model.to(lowerCamelCase__			)
           model.eval()
           with torch.no_grad():
              __UpperCamelCase	:			str					=					model(**self._prepare_for_class(lowerCamelCase__   ,				lowerCamelCase__			)			)
           __UpperCamelCase	:			str					=					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]   ,				)
           __UpperCamelCase	:			Optional[int]					=					len(lowerCamelCase__			)

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

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

           __UpperCamelCase	:			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    a					(    self						:	List[str]			):
     """simple docstring"""

     def check_hidden_states_output(lowerCamelCase__						:	str   ,				lowerCamelCase__						:	List[Any]   ,				lowerCamelCase__						:	Optional[int]			):
        __UpperCamelCase	:			Dict					=					model_class(lowerCamelCase__			)
        model.to(lowerCamelCase__			)
        model.eval()

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

        __UpperCamelCase	:			str					=					outputs.hidden_states
        __UpperCamelCase	:			List[str]					=					self.model_tester.num_hidden_layers + 1
        self.assertEqual(len(lowerCamelCase__			)   ,				lowerCamelCase__			)

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

     __UpperCamelCase	,     __UpperCamelCase	:			Optional[Any]					=					self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
        __UpperCamelCase	:			Optional[Any]					=					True
        check_hidden_states_output(lowerCamelCase__   ,				lowerCamelCase__   ,				lowerCamelCase__			)

        # check that output_hidden_states also work using config
        del inputs_dict["output_hidden_states"]
        __UpperCamelCase	:			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    a					(    self						:	str			):
     """simple docstring"""

     pass







def 				__lowerCamelCase     (							)     ->  List[str]:
   __UpperCamelCase	:			Optional[Any]					=					hf_hub_download(
       repo_id="""hf-internal-testing/spaghetti-video"""					,							filename="""eating_spaghetti.npy"""					,							repo_type="""dataset"""							)
   __UpperCamelCase	:			int					=					np.load(__lowerCAmelCase							)
   return list(__lowerCAmelCase							)







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

  @cached_property
  def    a					(    self						:	int			):
     """simple docstring"""

     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						:	int			):
     """simple docstring"""

     __UpperCamelCase	:			List[str]					=					VideoMAEForVideoClassification.from_pretrained("""MCG-NJU/videomae-base-finetuned-kinetics"""			).to(
         lowerCamelCase__			)

     __UpperCamelCase	:			Union[str, Any]					=					self.default_image_processor
     __UpperCamelCase	:			Optional[Any]					=					prepare_video()
     __UpperCamelCase	:			Optional[Any]					=					image_processor(lowerCamelCase__   ,				return_tensors="""pt"""			).to(lowerCamelCase__			)

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

     # verify the logits
     __UpperCamelCase	:			List[str]					=					torch.Size((1, 4_00)			)
     self.assertEqual(outputs.logits.shape   ,				lowerCamelCase__			)

     __UpperCamelCase	:			Tuple					=					torch.tensor([0.3669, -0.0688, -0.2421]			).to(lowerCamelCase__			)

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

  @slow
  def    a					(    self						:	Optional[int]			):
     """simple docstring"""

     __UpperCamelCase	:			Optional[Any]					=					VideoMAEForPreTraining.from_pretrained("""MCG-NJU/videomae-base-short"""			).to(lowerCamelCase__			)

     __UpperCamelCase	:			List[str]					=					self.default_image_processor
     __UpperCamelCase	:			List[Any]					=					prepare_video()
     __UpperCamelCase	:			List[Any]					=					image_processor(lowerCamelCase__   ,				return_tensors="""pt"""			).to(lowerCamelCase__			)

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

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

     # verify the logits
     __UpperCamelCase	:			Optional[Any]					=					torch.Size([1, 14_08, 15_36]			)
     __UpperCamelCase	:			List[Any]					=					torch.tensor(
         [[0.7994, 0.9612, 0.8508], [0.7401, 0.8958, 0.8302], [0.5862, 0.7468, 0.7325]]   ,				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`)
     __UpperCamelCase	:			Optional[Any]					=					torch.tensor([0.5142]   ,				device=lowerCamelCase__			)
     self.assertTrue(torch.allclose(outputs.loss   ,				lowerCamelCase__   ,				atol=1e-4			)			)

     # verify the loss (`config.norm_pix_loss` = `False`)
     __UpperCamelCase	:			List[str]					=					VideoMAEForPreTraining.from_pretrained("""MCG-NJU/videomae-base-short"""   ,				norm_pix_loss=lowerCamelCase__			).to(
         lowerCamelCase__			)

     with torch.no_grad():
        __UpperCamelCase	:			Dict					=					model(**lowerCamelCase__			)

     __UpperCamelCase	:			Optional[Any]					=					torch.tensor(torch.tensor([0.6469]			)   ,				device=lowerCamelCase__			)
     self.assertTrue(torch.allclose(outputs.loss   ,				lowerCamelCase__   ,				atol=1e-4			)			)











import os

import pytest

from transformers.dynamic_module_utils import get_imports


UpperCamelCase    							=      '\nimport os\n'

UpperCamelCase    							=      '\ndef foo():\n    import os\n    return False\n'

UpperCamelCase    							=      '\ndef foo():\n    def bar():\n        if True:\n            import os\n        return False\n    return bar()\n'

UpperCamelCase    							=      '\nimport os\n\ntry:\n    import bar\nexcept ImportError:\n    raise ValueError()\n'

UpperCamelCase    							=      '\nimport os\n\ndef foo():\n    try:\n        import bar\n    except ImportError:\n        raise ValueError()\n'

UpperCamelCase    							=      '\nimport os\n\ntry:\n    import bar\nexcept (ImportError, AttributeError):\n    raise ValueError()\n'

UpperCamelCase    							=      '\nimport os\n\ntry:\n    import bar\nexcept ImportError as e:\n    raise ValueError()\n'

UpperCamelCase    							=      '\nimport os\n\ntry:\n    import bar\nexcept:\n    raise ValueError()\n'

UpperCamelCase    							=      '\nimport os\n\ntry:\n    import bar\n    import baz\nexcept ImportError:\n    raise ValueError()\n'

UpperCamelCase    							=      '\nimport os\n\ntry:\n    import bar\n    import baz\nexcept ImportError:\n    x = 1\n    raise ValueError()\n'

UpperCamelCase    							=      [
    TOP_LEVEL_IMPORT,
    IMPORT_IN_FUNCTION,
    DEEPLY_NESTED_IMPORT,
    TOP_LEVEL_TRY_IMPORT,
    GENERIC_EXCEPT_IMPORT,
    MULTILINE_TRY_IMPORT,
    MULTILINE_BOTH_IMPORT,
    MULTIPLE_EXCEPTS_IMPORT,
    EXCEPT_AS_IMPORT,
    TRY_IMPORT_IN_FUNCTION,
]


@pytest.mark.parametrize("""case"""					,							__lowerCAmelCase							)
def 				__lowerCamelCase     (							__lowerCAmelCase							:					List[Any]					,							__lowerCAmelCase							:					Optional[Any]							)     ->  str:
   __UpperCamelCase	:			str					=					os.path.join(__lowerCAmelCase					,							"""test_file.py"""							)
   with open(__lowerCAmelCase					,							"""w"""							) as _tmp_file:
      _tmp_file.write(__lowerCAmelCase							)

   __UpperCamelCase	:			Optional[int]					=					get_imports(__lowerCAmelCase							)
   assert parsed_imports == ["os"]



