

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							:					int,		__A							:					Optional[int],		__A							:					Any=1_3,		__A							:					Union[str, Any]=1_0,		__A							:					Dict=3,		__A							:					int=2,		__A							:					Optional[int]=2,		__A							:					Any=2,		__A							:					Optional[int]=True,		__A							:					Optional[Any]=True,		__A							:					List[str]=3_2,		__A							:					int=5,		__A							:					List[Any]=4,		__A							:					Tuple=3_7,		__A							:					Dict="gelu",		__A							:					Any=0.1,		__A							:					Optional[Any]=0.1,		__A							:					int=1_0,		__A							:					Optional[Any]=0.0_2,		__A							:					Optional[Any]=0.9,		__A							:					Dict=None,		):
      UpperCAmelCase	:							int						=			parent
      UpperCAmelCase	:							Optional[Any]						=			batch_size
      UpperCAmelCase	:							Optional[Any]						=			image_size
      UpperCAmelCase	:							Optional[Any]						=			num_channels
      UpperCAmelCase	:							List[str]						=			patch_size
      UpperCAmelCase	:							int						=			tubelet_size
      UpperCAmelCase	:							Any						=			num_frames
      UpperCAmelCase	:							str						=			is_training
      UpperCAmelCase	:							Any						=			use_labels
      UpperCAmelCase	:							str						=			hidden_size
      UpperCAmelCase	:							int						=			num_hidden_layers
      UpperCAmelCase	:							Optional[int]						=			num_attention_heads
      UpperCAmelCase	:							Optional[Any]						=			intermediate_size
      UpperCAmelCase	:							Optional[Any]						=			hidden_act
      UpperCAmelCase	:							List[Any]						=			hidden_dropout_prob
      UpperCAmelCase	:							Optional[int]						=			attention_probs_dropout_prob
      UpperCAmelCase	:							List[Any]						=			type_sequence_label_size
      UpperCAmelCase	:							List[Any]						=			initializer_range
      UpperCAmelCase	:							Union[str, Any]						=			mask_ratio
      UpperCAmelCase	:							List[Any]						=			scope

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

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



    def 	__magic_name__   (							self							:					List[Any]						):
      UpperCAmelCase	:							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	:							Tuple						=			ids_tensor([self.batch_size],		self.type_sequence_label_size						)

      UpperCAmelCase	:							Union[str, Any]						=			self.get_config()

      return config, pixel_values, labels



    def 	__magic_name__   (							self							:					Union[str, 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=__snake_case,		initializer_range=self.initializer_range,		)



    def 	__magic_name__   (							self							:					List[str],		__A							:					List[str],		__A							:					List[str],		__A							:					Optional[Any]						):
      UpperCAmelCase	:							Optional[Any]						=			VideoMAEModel(config=__snake_case						)
      model.to(__snake_case						)
      model.eval()
      UpperCAmelCase	:							Union[str, Any]						=			model(__snake_case						)
      self.parent.assertEqual(result.last_hidden_state.shape,		(self.batch_size, self.seq_length, self.hidden_size)						)



    def 	__magic_name__   (							self							:					Optional[int],		__A							:					List[str],		__A							:					int,		__A							:					Union[str, Any]						):
      UpperCAmelCase	:							Union[str, Any]						=			VideoMAEForPreTraining(__snake_case						)
      model.to(__snake_case						)
      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	:							Dict						=			torch.ones((self.num_masks,)						)
      UpperCAmelCase	:							List[Any]						=			torch.cat([mask, torch.zeros(self.seq_length - mask.size(0						)						)]						)
      UpperCAmelCase	:							List[Any]						=			mask.expand(self.batch_size,		-1						).bool()

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





    def 	__magic_name__   (							self							:					Union[str, Any]						):
      UpperCAmelCase	:							Optional[int]						=			self.prepare_config_and_inputs()
      UpperCAmelCase			,						UpperCAmelCase			,						UpperCAmelCase	:							Union[str, Any]						=			config_and_inputs
      UpperCAmelCase	:							List[str]						=			{'''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 	__magic_name__   (							self							:					int						):
      UpperCAmelCase	:							List[str]						=			VideoMAEModelTester(self						)
      UpperCAmelCase	:							Optional[int]						=			ConfigTester(self,		config_class=__snake_case,		has_text_modality=__snake_case,		hidden_size=3_7						)



    def 	__magic_name__   (							self							:					List[str],		__A							:					Dict,		__A							:					Optional[Any],		__A							:					Optional[Any]=False						):
      UpperCAmelCase	:							Optional[Any]						=			copy.deepcopy(__snake_case						)

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

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

      return inputs_dict



    def 	__magic_name__   (							self							:					Any						):
      self.config_tester.run_common_tests()



    @unittest.skip(reason='''VideoMAE does not use inputs_embeds'''						)
    def 	__magic_name__   (							self							:					int						):
      pass



    def 	__magic_name__   (							self							:					Union[str, Any]						):
      UpperCAmelCase			,						UpperCAmelCase	:							str						=			self.model_tester.prepare_config_and_inputs_for_common()

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



    def 	__magic_name__   (							self							:					Union[str, Any]						):
      UpperCAmelCase			,						UpperCAmelCase	:							Dict						=			self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        UpperCAmelCase	:							List[Any]						=			model_class(__snake_case						)
        UpperCAmelCase	:							Union[str, Any]						=			inspect.signature(model.forward						)
        # signature.parameters is an OrderedDict => so arg_names order is deterministic
        UpperCAmelCase	:							Union[str, Any]						=			[*signature.parameters.keys()]

        UpperCAmelCase	:							List[str]						=			['''pixel_values''']
        self.assertListEqual(arg_names[:1],		__snake_case						)



    def 	__magic_name__   (							self							:					Optional[Any]						):
      UpperCAmelCase	:							Optional[Any]						=			self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*__snake_case						)



    def 	__magic_name__   (							self							:					Dict						):
      UpperCAmelCase	:							Union[str, Any]						=			self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_pretraining(*__snake_case						)



    @slow
    def 	__magic_name__   (							self							:					int						):
      for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        UpperCAmelCase	:							Optional[Any]						=			VideoMAEModel.from_pretrained(__snake_case						)
        self.assertIsNotNone(__snake_case						)



    def 	__magic_name__   (							self							:					Optional[int]						):
      if not self.has_attentions:
        pass

      else:
        UpperCAmelCase			,						UpperCAmelCase	:							Tuple						=			self.model_tester.prepare_config_and_inputs_for_common()
        UpperCAmelCase	:							List[Any]						=			True

        for model_class in self.all_model_classes:
          UpperCAmelCase	:							int						=			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
          )

          UpperCAmelCase	:							str						=			True
          UpperCAmelCase	:							Optional[int]						=			False
          UpperCAmelCase	:							Optional[int]						=			True
          UpperCAmelCase	:							List[str]						=			model_class(__snake_case						)
          model.to(__snake_case						)
          model.eval()
          with torch.no_grad():
            UpperCAmelCase	:							str						=			model(**self._prepare_for_class(__snake_case,		__snake_case						)						)
          UpperCAmelCase	:							Tuple						=			outputs.attentions
          self.assertEqual(len(__snake_case						),		self.model_tester.num_hidden_layers						)

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

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

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

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

          UpperCAmelCase	:							str						=			outputs.attentions

          self.assertEqual(len(__snake_case						),		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 	__magic_name__   (							self							:					Union[str, Any]						):
      def check_hidden_states_output(__A							:					Dict,		__A							:					Optional[int],		__A							:					Union[str, Any]						):
        UpperCAmelCase	:							str						=			model_class(__snake_case						)
        model.to(__snake_case						)
        model.eval()

        with torch.no_grad():
          UpperCAmelCase	:							str						=			model(**self._prepare_for_class(__snake_case,		__snake_case						)						)

        UpperCAmelCase	:							Optional[int]						=			outputs.hidden_states
        UpperCAmelCase	:							Dict						=			self.model_tester.num_hidden_layers + 1
        self.assertEqual(len(__snake_case						),		__snake_case						)

        UpperCAmelCase	:							Any						=			self.model_tester.seq_length - self.model_tester.num_masks
        UpperCAmelCase	:							Union[str, 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	:							Union[str, Any]						=			self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        UpperCAmelCase	:							List[str]						=			True
        check_hidden_states_output(__snake_case,		__snake_case,		__snake_case						)

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

        check_hidden_states_output(__snake_case,		__snake_case,		__snake_case						)





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






def       a__					(		)  ->						List[str]:
  UpperCAmelCase	:							List[Any]						=			hf_hub_download(
      repo_id='''hf-internal-testing/spaghetti-video'''    ,			filename='''eating_spaghetti.npy'''    ,			repo_type='''dataset'''      )
  UpperCAmelCase	:							Union[str, Any]						=			np.load(__lowerCAmelCase      )
  return list(__lowerCAmelCase      )






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



    @cached_property
    def 	__magic_name__   (							self							:					Optional[Any]						):
      # 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 	__magic_name__   (							self							:					Optional[int]						):
      UpperCAmelCase	:							Optional[int]						=			VideoMAEForVideoClassification.from_pretrained('''MCG-NJU/videomae-base-finetuned-kinetics'''						).to(
          __snake_case						)

      UpperCAmelCase	:							Any						=			self.default_image_processor
      UpperCAmelCase	:							Any						=			prepare_video()
      UpperCAmelCase	:							Any						=			image_processor(__snake_case,		return_tensors='''pt'''						).to(__snake_case						)

      # forward pass
      with torch.no_grad():
        UpperCAmelCase	:							Union[str, Any]						=			model(**__snake_case						)

      # verify the logits
      UpperCAmelCase	:							Dict						=			torch.Size((1, 4_0_0)						)
      self.assertEqual(outputs.logits.shape,		__snake_case						)

      UpperCAmelCase	:							List[Any]						=			torch.tensor([0.3_6_6_9, -0.0_6_8_8, -0.2_4_2_1]						).to(__snake_case						)

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





    @slow
    def 	__magic_name__   (							self							:					List[Any]						):
      UpperCAmelCase	:							str						=			VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''						).to(__snake_case						)

      UpperCAmelCase	:							Optional[Any]						=			self.default_image_processor
      UpperCAmelCase	:							str						=			prepare_video()
      UpperCAmelCase	:							int						=			image_processor(__snake_case,		return_tensors='''pt'''						).to(__snake_case						)

      # add boolean mask, indicating which patches to mask
      UpperCAmelCase	:							Union[str, Any]						=			hf_hub_download(repo_id='''hf-internal-testing/bool-masked-pos''',		filename='''bool_masked_pos.pt'''						)
      UpperCAmelCase	:							List[str]						=			torch.load(__snake_case						)

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

      # verify the logits
      UpperCAmelCase	:							Dict						=			torch.Size([1, 1_4_0_8, 1_5_3_6]						)
      UpperCAmelCase	:							str						=			torch.tensor(
          [[0.7_9_9_4, 0.9_6_1_2, 0.8_5_0_8], [0.7_4_0_1, 0.8_9_5_8, 0.8_3_0_2], [0.5_8_6_2, 0.7_4_6_8, 0.7_3_2_5]],		device=__snake_case						)
      self.assertEqual(outputs.logits.shape,		__snake_case						)
      self.assertTrue(torch.allclose(outputs.logits[0, :3, :3],		__snake_case,		atol=1E-4						)						)

      # verify the loss (`config.norm_pix_loss` = `True`)
      UpperCAmelCase	:							Any						=			torch.tensor([0.5_1_4_2],		device=__snake_case						)
      self.assertTrue(torch.allclose(outputs.loss,		__snake_case,		atol=1E-4						)						)

      # verify the loss (`config.norm_pix_loss` = `False`)
      UpperCAmelCase	:							Optional[int]						=			VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short''',		norm_pix_loss=__snake_case						).to(
          __snake_case						)

      with torch.no_grad():
        UpperCAmelCase	:							Optional[int]						=			model(**__snake_case						)

      UpperCAmelCase	:							List[Any]						=			torch.tensor(torch.tensor([0.6_4_6_9]						),		device=__snake_case						)
      self.assertTrue(torch.allclose(outputs.loss,		__snake_case,		atol=1E-4						)						)








from collections.abc import Callable
from math import pi, sqrt
from random import uniform
from statistics import mean

def       a__					(		UpperCAmelCase							:      int      )  ->						Dict:

  # A local function to see if a dot lands in the circle.
  def is_in_circle(UpperCAmelCase							:      float    ,			UpperCAmelCase							:      float      ) -> bool:
    UpperCAmelCase	:							Dict						=			sqrt((x**2) + (y**2)      )
    # Our circle has a radius of 1, so a distance
    # greater than 1 would land outside the circle.
    return distance_from_centre <= 1

  # The proportion of guesses that landed in the circle
  UpperCAmelCase	:							List[str]						=			mean(
      int(is_in_circle(uniform(-1.0    ,			1.0      )    ,			uniform(-1.0    ,			1.0      )      )      )
      for _ in range(UpperCAmelCase      )      )
  # The ratio of the area for circle to square is pi/4.
  UpperCAmelCase	:							int						=			proportion * 4
  print(f'''The estimated value of pi is {pi_estimate}'''      )
  print(f'''The numpy value of pi is {pi}'''      )
  print(f'''The total error is {abs(pi - pi_estimate      )}'''      )

def       a__					(		UpperCAmelCase							:      int    ,			UpperCAmelCase							:      Callable[[float], float]    ,			UpperCAmelCase							:      float = 0.0    ,			UpperCAmelCase							:      float = 1.0    ,			)  ->						float:
  return mean(
      function_to_integrate(uniform(UpperCAmelCase    ,			UpperCAmelCase      )      ) for _ in range(UpperCAmelCase      )      ) * (max_value - min_value)

def       a__					(		UpperCAmelCase							:      int    ,			UpperCAmelCase							:      float = 0.0    ,			UpperCAmelCase							:      float = 1.0      )  ->						None:

  def identity_function(UpperCAmelCase							:      float      ) -> float:
    return x

  UpperCAmelCase	:							int						=			area_under_curve_estimator(
      UpperCAmelCase    ,			UpperCAmelCase    ,			UpperCAmelCase    ,			UpperCAmelCase      )
  UpperCAmelCase	:							Tuple						=			(max_value * max_value - min_value * min_value) / 2

  print('''******************'''      )
  print(f'''Estimating area under y=x where x varies from {min_value} to {max_value}'''      )
  print(f'''Estimated value is {estimated_value}'''      )
  print(f'''Expected value is {expected_value}'''      )
  print(f'''Total error is {abs(estimated_value - expected_value      )}'''      )
  print('''******************'''      )

def       a__					(		UpperCAmelCase							:      int      )  ->						None:

  def function_to_integrate(UpperCAmelCase							:      float      ) -> float:
    return sqrt(4.0 - x * x      )

  UpperCAmelCase	:							Optional[int]						=			area_under_curve_estimator(
      UpperCAmelCase    ,			UpperCAmelCase    ,			0.0    ,			2.0      )

  print('''******************'''      )
  print('''Estimating pi using area_under_curve_estimator'''      )
  print(f'''Estimated value is {estimated_value}'''      )
  print(f'''Expected value is {pi}'''      )
  print(f'''Total error is {abs(estimated_value - pi      )}'''      )
  print('''******************'''      )


if __name__ == "__main__":
     import doctest

     doctest.testmod()





