


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




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

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

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




 def        SCREAMING_SNAKE_CASE__				(		self    ):
       a						:Any 							=       floats_tensor(
           [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]    )

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

       a						:int 							=       self.get_config()

       return config, pixel_values, labels




 def        SCREAMING_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        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase    ):
       a						:List[str] 							=       VideoMAEModel(config=_lowerCamelCase    )
       model.to(_lowerCamelCase    )
       model.eval()
       a						:Dict 							=       model(_lowerCamelCase    )
       self.parent.assertEqual(result.last_hidden_state.shape					,  (self.batch_size, self.seq_length, self.hidden_size)    )




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase    ):
       a						:Optional[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
       a						:Optional[Any] 							=       torch.ones((self.num_masks,)    )
       a						:str 							=       torch.cat([mask, torch.zeros(self.seq_length - mask.size(0    )    )]    )
       a						:List[str] 							=       mask.expand(self.batch_size					,  -1    ).bool()

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




 def        SCREAMING_SNAKE_CASE__				(		self    ):
       a						:str 							=       self.prepare_config_and_inputs()
       a							,				a							,				a						:Union[str, Any] 							=       config_and_inputs
       a						:Union[str, Any] 							=       {'''pixel_values''': pixel_values}
       return config, inputs_dict







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

 SCREAMING_SNAKE_CASE__      =       False
 SCREAMING_SNAKE_CASE__      =       False
 SCREAMING_SNAKE_CASE__      =       False
 SCREAMING_SNAKE_CASE__      =       False




 def        SCREAMING_SNAKE_CASE__				(		self    ):
       a						:List[Any] 							=       VideoMAEModelTester(self    )
       a						:Any 							=       ConfigTester(self					,  config_class=_lowerCamelCase					,  has_text_modality=_lowerCamelCase					,  hidden_size=37    )




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

       if return_labels:
             if model_class in [
                 *get_values(_lowerCamelCase    ),
             ]:
                   a						:List[Any] 							=       torch.zeros(
                       self.model_tester.batch_size					,  dtype=torch.long					,  device=_lowerCamelCase    )

       return inputs_dict




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




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




 def        SCREAMING_SNAKE_CASE__				(		self    ):
       a							,				a						:Dict 							=       self.model_tester.prepare_config_and_inputs_for_common()

       for model_class in self.all_model_classes:
             a						:Optional[int] 							=       model_class(_lowerCamelCase    )
             self.assertIsInstance(model.get_input_embeddings()					,  (nn.Module)    )
             a						:Dict 							=       model.get_output_embeddings()
             self.assertTrue(x is None or isinstance(_lowerCamelCase					,  nn.Linear    )    )




 def        SCREAMING_SNAKE_CASE__				(		self    ):
       a							,				a						:str 							=       self.model_tester.prepare_config_and_inputs_for_common()

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

             a						:str 							=       ['''pixel_values''']
             self.assertListEqual(arg_names[:1]					,  _lowerCamelCase    )




 def        SCREAMING_SNAKE_CASE__				(		self    ):
       a						:Optional[int] 							=       self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_model(*_lowerCamelCase    )




 def        SCREAMING_SNAKE_CASE__				(		self    ):
       a						:Tuple 							=       self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_pretraining(*_lowerCamelCase    )




 @slow
 def        SCREAMING_SNAKE_CASE__				(		self    ):
       for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
             a						:Optional[Any] 							=       VideoMAEModel.from_pretrained(_lowerCamelCase    )
             self.assertIsNotNone(_lowerCamelCase    )




 def        SCREAMING_SNAKE_CASE__				(		self    ):
       if not self.has_attentions:
             pass

       else:
             a							,				a						:Optional[int] 							=       self.model_tester.prepare_config_and_inputs_for_common()
             a						:List[Any] 							=       True

             for model_class in self.all_model_classes:
                   a						:Union[str, Any] 							=       self.model_tester.seq_length - self.model_tester.num_masks
                   a						:List[Any] 							=       (
                       num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length
                   )

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

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

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

                   a						:int 							=       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        SCREAMING_SNAKE_CASE__				(		self    ):
       def check_hidden_states_output(_lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase    ):
             a						:Optional[Any] 							=       model_class(_lowerCamelCase    )
             model.to(_lowerCamelCase    )
             model.eval()

             with torch.no_grad():
                   a						:List[Any] 							=       model(**self._prepare_for_class(_lowerCamelCase					,  _lowerCamelCase    )    )

             a						:List[Any] 							=       outputs.hidden_states
             a						:int 							=       self.model_tester.num_hidden_layers + 1
             self.assertEqual(len(_lowerCamelCase    )					,  _lowerCamelCase    )

             a						:int 							=       self.model_tester.seq_length - self.model_tester.num_masks
             a						:Dict 							=       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]					,  )

       a							,				a						:Optional[int] 							=       self.model_tester.prepare_config_and_inputs_for_common()

       for model_class in self.all_model_classes:
             a						: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"]
             a						:Tuple 							=       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        SCREAMING_SNAKE_CASE__				(		self    ):
       pass







def 				__lowerCamelCase						( ):




      """simple docstring"""





      a						:Optional[Any] 							=       hf_hub_download(
          repo_id='''hf-internal-testing/spaghetti-video''' ,  filename='''eating_spaghetti.npy''' ,  repo_type='''dataset'''						)
      a						:Optional[Any] 							=       np.load(UpperCAmelCase_						)
      return list(UpperCAmelCase_						)







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




 @cached_property
 def        SCREAMING_SNAKE_CASE__				(		self    ):
       # 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        SCREAMING_SNAKE_CASE__				(		self    ):
       a						:List[Any] 							=       VideoMAEForVideoClassification.from_pretrained('''MCG-NJU/videomae-base-finetuned-kinetics'''    ).to(
           _lowerCamelCase    )

       a						:Tuple 							=       self.default_image_processor
       a						:Optional[int] 							=       prepare_video()
       a						:str 							=       image_processor(_lowerCamelCase					,  return_tensors='''pt'''    ).to(_lowerCamelCase    )

       # forward pass
       with torch.no_grad():
             a						:Tuple 							=       model(**_lowerCamelCase    )

       # verify the logits
       a						:List[str] 							=       torch.Size((1, 400)    )
       self.assertEqual(outputs.logits.shape					,  _lowerCamelCase    )

       a						:Optional[int] 							=       torch.tensor([0.3669, -0.0688, -0.2421]    ).to(_lowerCamelCase    )

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




 @slow
 def        SCREAMING_SNAKE_CASE__				(		self    ):
       a						:str 							=       VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''    ).to(_lowerCamelCase    )

       a						:str 							=       self.default_image_processor
       a						:List[str] 							=       prepare_video()
       a						:Union[str, Any] 							=       image_processor(_lowerCamelCase					,  return_tensors='''pt'''    ).to(_lowerCamelCase    )

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

       # forward pass
       with torch.no_grad():
             a						:Optional[Any] 							=       model(**_lowerCamelCase    )

       # verify the logits
       a						:List[Any] 							=       torch.Size([1, 1408, 1536]    )
       a						:Union[str, 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`)
       a						:Union[str, 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`)
       a						:List[str] 							=       VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''					,  norm_pix_loss=_lowerCamelCase    ).to(
           _lowerCamelCase    )

       with torch.no_grad():
             a						:Tuple 							=       model(**_lowerCamelCase    )

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









"""simple docstring"""


class   _UpperCamelCase  (   lowerCAmelCase__		):
							'''simple docstring'''







							pass
class   _UpperCamelCase  (   lowerCAmelCase__		):
							'''simple docstring'''







							pass
class   _UpperCamelCase  :
							'''simple docstring'''







							def __init__(     self	):
										__lowerCAmelCase				    =			[
										    [],
										    [],
										    [],
										]

							def 			snake_case				(     self    ,   __a    ,   __a	):
										try:
													if len(self.queues[priority]	) >= 1_00:
																raise OverflowError("Maximum queue size is 100"	)
													self.queues[priority].append(__a	)
										except IndexError:
													raise ValueError("Valid priorities are 0, 1, and 2"	)

							def 			snake_case				(     self	):
										for queue in self.queues:
													if queue:
																return queue.pop(0	)
										raise UnderFlowError("All queues are empty"	)



							def __str__(     self	):
										return "\n".join(f"Priority {i}: {q}" for i, q in enumerate(self.queues	)	)
class   _UpperCamelCase  :
							'''simple docstring'''







							def __init__(     self	):
										__lowerCAmelCase				    =			[]

							def 			snake_case				(     self    ,   __a	):
										if len(self.queue	) == 1_00:
													raise OverFlowError("Maximum queue size is 100"	)
										self.queue.append(__a	)

							def 			snake_case				(     self	):
										if not self.queue:
													raise UnderFlowError("The queue is empty"	)
										else:
													__lowerCAmelCase				    =			min(self.queue	)
													self.queue.remove(__a	)
													return data



							def __str__(     self	):
										return str(self.queue	)


def _lowerCamelCase	(		):

			'''simple docstring'''




			__lowerCAmelCase				    =			FixedPriorityQueue()
			fpq.enqueue(0    , 10      )
			fpq.enqueue(1    , 70      )
			fpq.enqueue(0    , 100      )
			fpq.enqueue(2    , 1      )
			fpq.enqueue(2    , 5      )
			fpq.enqueue(1    , 7      )
			fpq.enqueue(2    , 4      )
			fpq.enqueue(1    , 64      )
			fpq.enqueue(0    , 128      )
			print(_UpperCamelCase      )
			print(fpq.dequeue()      )
			print(fpq.dequeue()      )
			print(fpq.dequeue()      )
			print(fpq.dequeue()      )
			print(fpq.dequeue()      )
			print(_UpperCamelCase      )
			print(fpq.dequeue()      )
			print(fpq.dequeue()      )
			print(fpq.dequeue()      )
			print(fpq.dequeue()      )
			print(fpq.dequeue()      )


def _lowerCamelCase	(		):

			'''simple docstring'''




			__lowerCAmelCase				    =			ElementPriorityQueue()
			epq.enqueue(10      )
			epq.enqueue(70      )
			epq.enqueue(100      )
			epq.enqueue(1      )
			epq.enqueue(5      )
			epq.enqueue(7      )
			epq.enqueue(4      )
			epq.enqueue(64      )
			epq.enqueue(128      )
			print(_UpperCamelCase      )
			print(epq.dequeue()      )
			print(epq.dequeue()      )
			print(epq.dequeue()      )
			print(epq.dequeue()      )
			print(epq.dequeue()      )
			print(_UpperCamelCase      )
			print(epq.dequeue()      )
			print(epq.dequeue()      )
			print(epq.dequeue()      )
			print(epq.dequeue()      )
			print(epq.dequeue()      )


if __name__ == "__main__":
	fixed_priority_queue()
	element_priority_queue()




