






"""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     :						Union[str, Any]    ,						A     :						Tuple    ,						A     :						Optional[Any]=13    ,						A     :						int=10    ,						A     :						Dict=3    ,						A     :						Union[str, Any]=2    ,						A     :						Dict=2    ,						A     :						Tuple=2    ,						A     :						str=True    ,						A     :						str=True    ,						A     :						List[str]=32    ,						A     :						Optional[int]=5    ,						A     :						Any=4    ,						A     :						Dict=37    ,						A     :						Optional[int]="gelu"    ,						A     :						List[Any]=0.1    ,						A     :						List[str]=0.1    ,						A     :						Optional[Any]=10    ,						A     :						Optional[int]=0.02    ,						A     :						Any=0.9    ,						A     :						List[Any]=None    ,						)  ->  Any:
							lowercase_ : Optional[int]							       =   parent
							lowercase_ : Dict							       =   batch_size
							lowercase_ : Optional[int]							       =   image_size
							lowercase_ : Optional[Any]							       =   num_channels
							lowercase_ : Tuple							       =   patch_size
							lowercase_ : int							       =   tubelet_size
							lowercase_ : Union[str, Any]							       =   num_frames
							lowercase_ : List[str]							       =   is_training
							lowercase_ : List[Any]							       =   use_labels
							lowercase_ : List[str]							       =   hidden_size
							lowercase_ : Dict							       =   num_hidden_layers
							lowercase_ : Any							       =   num_attention_heads
							lowercase_ : Any							       =   intermediate_size
							lowercase_ : Any							       =   hidden_act
							lowercase_ : Optional[Any]							       =   hidden_dropout_prob
							lowercase_ : Tuple							       =   attention_probs_dropout_prob
							lowercase_ : Optional[int]							       =   type_sequence_label_size
							lowercase_ : int							       =   initializer_range
							lowercase_ : List[str]							       =   mask_ratio
							lowercase_ : Optional[Any]							       =   scope

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

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




					def   A  (				self     :						str   )  ->  str:
							lowercase_ : List[Any]							       =   floats_tensor(
							    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]   )

							lowercase_ : Optional[int]							       =   None
							if self.use_labels:
									lowercase_ : Optional[int]							       =   ids_tensor([self.batch_size]    ,						self.type_sequence_label_size   )

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

							return config, pixel_values, labels




					def   A  (				self     :						Optional[Any]   )  ->  List[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     :						Optional[Any]    ,						A     :						Optional[int]    ,						A     :						str    ,						A     :						Optional[int]   )  ->  Any:
							lowercase_ : Union[str, Any]							       =   VideoMAEModel(config=A   )
							model.to(A   )
							model.eval()
							lowercase_ : Optional[int]							       =   model(A   )
							self.parent.assertEqual(result.last_hidden_state.shape    ,						(self.batch_size, self.seq_length, self.hidden_size)   )




					def   A  (				self     :						Dict    ,						A     :						str    ,						A     :						str    ,						A     :						Any   )  ->  Any:
							lowercase_ : List[Any]							       =   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
							lowercase_ : Union[str, Any]							       =   torch.ones((self.num_masks,)   )
							lowercase_ : Tuple							       =   torch.cat([mask, torch.zeros(self.seq_length - mask.size(0   )   )]   )
							lowercase_ : List[Any]							       =   mask.expand(self.batch_size    ,						-1   ).bool()

							lowercase_ : Tuple							       =   model(A    ,						A   )
							# model only returns predictions for masked patches
							lowercase_ : List[str]							       =   mask.sum().item()
							lowercase_ : 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   A  (				self     :						List[Any]   )  ->  Any:
							lowercase_ : Union[str, Any]							       =   self.prepare_config_and_inputs()
							lowercase_       ,					lowercase_       ,					lowercase_ : str							       =   config_and_inputs
							lowercase_ : Union[str, Any]							       =   {'''pixel_values''': pixel_values}
							return config, inputs_dict





@require_torch
class 					_UpperCAmelCase       ( _A   ,						_A   ,						unittest.TestCase							):
					SCREAMING_SNAKE_CASE_		:				List[str]							    =   (
					    (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
					)
					SCREAMING_SNAKE_CASE_		:				str							    =   (
					    {"feature-extraction": VideoMAEModel, "video-classification": VideoMAEForVideoClassification}
					    if is_torch_available()
					    else {}
					)

					SCREAMING_SNAKE_CASE_		:				int							    =   False
					SCREAMING_SNAKE_CASE_		:				str							    =   False
					SCREAMING_SNAKE_CASE_		:				Any							    =   False
					SCREAMING_SNAKE_CASE_		:				List[str]							    =   False




					def   A  (				self     :						Any   )  ->  int:
							lowercase_ : int							       =   VideoMAEModelTester(self   )
							lowercase_ : str							       =   ConfigTester(self    ,						config_class=A    ,						has_text_modality=A    ,						hidden_size=37   )




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

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

							return inputs_dict




					def   A  (				self     :						Any   )  ->  Optional[int]:
							self.config_tester.run_common_tests()




					@unittest.skip(reason='''VideoMAE does not use inputs_embeds'''   )
					def   A  (				self     :						Dict   )  ->  List[Any]:
							pass




					def   A  (				self     :						Tuple   )  ->  str:
							lowercase_       ,					lowercase_ : int							       =   self.model_tester.prepare_config_and_inputs_for_common()

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




					def   A  (				self     :						Union[str, Any]   )  ->  Optional[Any]:
							lowercase_       ,					lowercase_ : Tuple							       =   self.model_tester.prepare_config_and_inputs_for_common()

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

									lowercase_ : Dict							       =   ['''pixel_values''']
									self.assertListEqual(arg_names[:1]    ,						A   )




					def   A  (				self     :						Tuple   )  ->  List[str]:
							lowercase_ : int							       =   self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_model(*A   )




					def   A  (				self     :						Dict   )  ->  int:
							lowercase_ : str							       =   self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_pretraining(*A   )




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




					def   A  (				self     :						Union[str, Any]   )  ->  int:
							if not self.has_attentions:
									pass

							else:
									lowercase_       ,					lowercase_ : Tuple							       =   self.model_tester.prepare_config_and_inputs_for_common()
									lowercase_ : List[str]							       =   True

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

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

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

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

											lowercase_ : Union[str, Any]							       =   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     :						int   )  ->  List[str]:
							def check_hidden_states_output(A     :						Dict    ,						A     :						List[Any]    ,						A     :						Optional[Any]   ):
									lowercase_ : Tuple							       =   model_class(A   )
									model.to(A   )
									model.eval()

									with torch.no_grad():
											lowercase_ : Union[str, Any]							       =   model(**self._prepare_for_class(A    ,						A   )   )

									lowercase_ : Any							       =   outputs.hidden_states
									lowercase_ : Any							       =   self.model_tester.num_hidden_layers + 1
									self.assertEqual(len(A   )    ,						A   )

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

							lowercase_       ,					lowercase_ : Tuple							       =   self.model_tester.prepare_config_and_inputs_for_common()

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

									# check that output_hidden_states also work using config
									del inputs_dict["output_hidden_states"]
									lowercase_ : List[Any]							       =   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     :						Tuple   )  ->  Tuple:
							pass



def 						lowercase    (				):
		lowercase_ : Optional[int]							       =   hf_hub_download(
		    repo_id='''hf-internal-testing/spaghetti-video'''		,					filename='''eating_spaghetti.npy'''		,					repo_type='''dataset'''	)
		lowercase_ : str							       =   np.load(__snake_case	)
		return list(__snake_case	)






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




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

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

							# forward pass
							with torch.no_grad():
									lowercase_ : List[str]							       =   model(**A   )

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

							lowercase_ : Union[str, Any]							       =   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[Any]   )  ->  List[Any]:
							lowercase_ : int							       =   VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''   ).to(A   )

							lowercase_ : Any							       =   self.default_image_processor
							lowercase_ : Union[str, Any]							       =   prepare_video()
							lowercase_ : Tuple							       =   image_processor(A    ,						return_tensors='''pt'''   ).to(A   )

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

							# forward pass
							with torch.no_grad():
									lowercase_ : List[Any]							       =   model(**A   )

							# verify the logits
							lowercase_ : Tuple							       =   torch.Size([1, 14_08, 15_36]   )
							lowercase_ : Tuple							       =   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`)
							lowercase_ : Any							       =   torch.tensor([0.5142]    ,						device=A   )
							self.assertTrue(torch.allclose(outputs.loss    ,						A    ,						atol=1e-4   )   )

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

							with torch.no_grad():
									lowercase_ : int							       =   model(**A   )

							lowercase_ : Dict							       =   torch.tensor(torch.tensor([0.6469]   )    ,						device=A   )
							self.assertTrue(torch.allclose(outputs.loss    ,						A    ,						atol=1e-4   )   )




"""simple docstring"""
from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowercase__					       =	logging.get_logger(__name__)

lowercase__					       =	{
    'edbeeching/decision-transformer-gym-hopper-medium': (
        'https://huggingface.co/edbeeching/decision-transformer-gym-hopper-medium/resolve/main/config.json'
    ),
    # See all DecisionTransformer models at https://huggingface.co/models?filter=decision_transformer
}




class    __snake_case       (     __lowerCAmelCase				):
      a__        =		"""decision_transformer"""
      a__        =		["""past_key_values"""]
      a__        =		{
          """max_position_embeddings""": """n_positions""",
          """num_attention_heads""": """n_head""",
          """num_hidden_layers""": """n_layer""",
      }



      def __init__(						self    ,			lowercase=17    ,			lowercase=4    ,			lowercase=1_28    ,			lowercase=40_96    ,			lowercase=True    ,			lowercase=1    ,			lowercase=10_24    ,			lowercase=3    ,			lowercase=1    ,			lowercase=None    ,			lowercase="relu"    ,			lowercase=0.1    ,			lowercase=0.1    ,			lowercase=0.1    ,			lowercase=1e-5    ,			lowercase=0.02    ,			lowercase=True    ,			lowercase=True    ,			lowercase=5_02_56    ,			lowercase=5_02_56    ,			lowercase=False    ,			lowercase=False    ,			**lowercase    ,			)	->			Tuple:





        '''simple docstring'''





        a__:     List[str]										=	state_dim
        a__:     int										=	act_dim
        a__:     List[Any]										=	hidden_size
        a__:     List[str]										=	max_ep_len
        a__:     List[Any]										=	action_tanh
        a__:     Optional[Any]										=	vocab_size
        a__:     Tuple										=	n_positions
        a__:     Dict										=	n_layer
        a__:     Optional[int]										=	n_head
        a__:     Optional[int]										=	n_inner
        a__:     Any										=	activation_function
        a__:     Union[str, Any]										=	resid_pdrop
        a__:     Any										=	embd_pdrop
        a__:     Any										=	attn_pdrop
        a__:     List[Any]										=	layer_norm_epsilon
        a__:     Optional[Any]										=	initializer_range
        a__:     Any										=	scale_attn_weights
        a__:     Dict										=	use_cache
        a__:     Optional[int]										=	scale_attn_by_inverse_layer_idx
        a__:     List[str]										=	reorder_and_upcast_attn

        a__:     Any										=	bos_token_id
        a__:     int										=	eos_token_id

        super().__init__(bos_token_id=lowercase    ,			eos_token_id=lowercase    ,			**lowercase)


