

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





    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						,       ):
         lowercase__   :       Tuple				=	parent
         lowercase__   :       int				=	batch_size
         lowercase__   :       Any				=	image_size
         lowercase__   :       Optional[int]				=	num_channels
         lowercase__   :       str				=	patch_size
         lowercase__   :       Any				=	tubelet_size
         lowercase__   :       Optional[int]				=	num_frames
         lowercase__   :       str				=	is_training
         lowercase__   :       Dict				=	use_labels
         lowercase__   :       str				=	hidden_size
         lowercase__   :       Dict				=	num_hidden_layers
         lowercase__   :       Any				=	num_attention_heads
         lowercase__   :       int				=	intermediate_size
         lowercase__   :       str				=	hidden_act
         lowercase__   :       Any				=	hidden_dropout_prob
         lowercase__   :       int				=	attention_probs_dropout_prob
         lowercase__   :       Optional[Any]				=	type_sequence_label_size
         lowercase__   :       Optional[int]				=	initializer_range
         lowercase__   :       Tuple				=	mask_ratio
         lowercase__   :       int				=	scope

         # in VideoMAE, the number of tokens equals num_frames/tubelet_size * num_patches per frame
         lowercase__   :       int				=	(image_size // patch_size) ** 2
         lowercase__   :       List[Any]				=	(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 				snake_case_    (				self):
         lowercase__   :       Any				=	floats_tensor(
             [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size])

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

         lowercase__   :       str				=	self.get_config()

         return config, pixel_values, labels





    def 				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=a						,       initializer_range=self.initializer_range						,       )





    def 				snake_case_    (				self						,       a						,       a						,       a):
         lowercase__   :       Dict				=	VideoMAEModel(config=a)
         model.to(a)
         model.eval()
         lowercase__   :       List[str]				=	model(a)
         self.parent.assertEqual(result.last_hidden_state.shape						,       (self.batch_size, self.seq_length, self.hidden_size))





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

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





    def 				snake_case_    (				self):
         lowercase__   :       Tuple				=	self.prepare_config_and_inputs()
         lowercase__				,			lowercase__				,			lowercase__   :       Union[str, Any]				=	config_and_inputs
         lowercase__   :       List[Any]				=	{'pixel_values': pixel_values}
         return config, inputs_dict



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

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





    def 				snake_case_    (				self):
         lowercase__   :       Dict				=	VideoMAEModelTester(self)
         lowercase__   :       int				=	ConfigTester(self						,       config_class=a						,       has_text_modality=a						,       hidden_size=37)





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

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

         return inputs_dict





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





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





    def 				snake_case_    (				self):
         lowercase__				,			lowercase__   :       Optional[int]				=	self.model_tester.prepare_config_and_inputs_for_common()

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





    def 				snake_case_    (				self):
         lowercase__				,			lowercase__   :       Tuple				=	self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
              lowercase__   :       int				=	model_class(a)
              lowercase__   :       Optional[int]				=	inspect.signature(model.forward)
              # signature.parameters is an OrderedDict => so arg_names order is deterministic
              lowercase__   :       List[str]				=	[*signature.parameters.keys()]

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





    def 				snake_case_    (				self):
         lowercase__   :       Optional[Any]				=	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*a)





    def 				snake_case_    (				self):
         lowercase__   :       Optional[Any]				=	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_pretraining(*a)





    @slow
    def 				snake_case_    (				self):
         for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
              lowercase__   :       str				=	VideoMAEModel.from_pretrained(a)
              self.assertIsNotNone(a)





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

         else:
              lowercase__				,			lowercase__   :       Any				=	self.model_tester.prepare_config_and_inputs_for_common()
              lowercase__   :       List[str]				=	True

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

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

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

                   lowercase__   :       Tuple				=	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 				snake_case_    (				self):
         def check_hidden_states_output(a						,       a						,       a):
              lowercase__   :       int				=	model_class(a)
              model.to(a)
              model.eval()

              with torch.no_grad():
                   lowercase__   :       int				=	model(**self._prepare_for_class(a						,       a))

              lowercase__   :       int				=	outputs.hidden_states
              lowercase__   :       str				=	self.model_tester.num_hidden_layers + 1
              self.assertEqual(len(a)						,       a)

              lowercase__   :       List[Any]				=	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

              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[Any]				=	True
              check_hidden_states_output(a						,       a						,       a)

              # check that output_hidden_states also work using config
              del inputs_dict["output_hidden_states"]
              lowercase__   :       str				=	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 				snake_case_    (				self):
         pass



def      snake_case__      (   ):



     '''simple docstring'''





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


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





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

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

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

         # verify the logits
         lowercase__   :       Any				=	torch.Size((1, 400))
         self.assertEqual(outputs.logits.shape						,       a)

         lowercase__   :       str				=	torch.tensor([0.3_669, -0.0_688, -0.2_421]).to(a)

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





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

         lowercase__   :       List[Any]				=	self.default_image_processor
         lowercase__   :       Optional[int]				=	prepare_video()
         lowercase__   :       Any				=	image_processor(a						,       return_tensors='pt').to(a)

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

         # forward pass
         with torch.no_grad():
              lowercase__   :       str				=	model(**a)

         # verify the logits
         lowercase__   :       List[Any]				=	torch.Size([1, 1408, 1536])
         lowercase__   :       Tuple				=	torch.tensor(
             [[0.7_994, 0.9_612, 0.8_508], [0.7_401, 0.8_958, 0.8_302], [0.5_862, 0.7_468, 0.7_325]]						,       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__   :       Union[str, Any]				=	torch.tensor([0.5_142]						,       device=a)
         self.assertTrue(torch.allclose(outputs.loss						,       a						,       atol=1e-4))

         # verify the loss (`config.norm_pix_loss` = `False`)
         lowercase__   :       List[Any]				=	VideoMAEForPreTraining.from_pretrained('MCG-NJU/videomae-base-short'						,       norm_pix_loss=a).to(
             a)

         with torch.no_grad():
              lowercase__   :       Any				=	model(**a)

         lowercase__   :       Optional[int]				=	torch.tensor(torch.tensor([0.6_469])						,       device=a)
         self.assertTrue(torch.allclose(outputs.loss						,       a						,       atol=1e-4))




import json
from typing import List, Optional, Tuple

from tokenizers import normalizers
from tokenizers.pre_tokenizers import BertPreTokenizer, PreTokenizer

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_roformer import RoFormerTokenizer
from .tokenization_utils import JiebaPreTokenizer


snake_case_        =			logging.get_logger(__name__)

snake_case_        =			{'''vocab_file''': '''vocab.txt''', '''tokenizer_file''': '''tokenizer.json'''}

snake_case_        =			{
    '''vocab_file''': {
        '''junnyu/roformer_chinese_small''': '''https://huggingface.co/junnyu/roformer_chinese_small/resolve/main/vocab.txt''',
        '''junnyu/roformer_chinese_base''': '''https://huggingface.co/junnyu/roformer_chinese_base/resolve/main/vocab.txt''',
        '''junnyu/roformer_chinese_char_small''': (
            '''https://huggingface.co/junnyu/roformer_chinese_char_small/resolve/main/vocab.txt'''
        ),
        '''junnyu/roformer_chinese_char_base''': (
            '''https://huggingface.co/junnyu/roformer_chinese_char_base/resolve/main/vocab.txt'''
        ),
        '''junnyu/roformer_small_discriminator''': (
            '''https://huggingface.co/junnyu/roformer_small_discriminator/resolve/main/vocab.txt'''
        ),
        '''junnyu/roformer_small_generator''': (
            '''https://huggingface.co/junnyu/roformer_small_generator/resolve/main/vocab.txt'''
        ),
    }
}

snake_case_        =			{
    '''junnyu/roformer_chinese_small''': 1_536,
    '''junnyu/roformer_chinese_base''': 1_536,
    '''junnyu/roformer_chinese_char_small''': 512,
    '''junnyu/roformer_chinese_char_base''': 512,
    '''junnyu/roformer_small_discriminator''': 128,
    '''junnyu/roformer_small_generator''': 128,
}


snake_case_        =			{
    '''junnyu/roformer_chinese_small''': {'''do_lower_case''': True},
    '''junnyu/roformer_chinese_base''': {'''do_lower_case''': True},
    '''junnyu/roformer_chinese_char_small''': {'''do_lower_case''': True},
    '''junnyu/roformer_chinese_char_base''': {'''do_lower_case''': True},
    '''junnyu/roformer_small_discriminator''': {'''do_lower_case''': True},
    '''junnyu/roformer_small_generator''': {'''do_lower_case''': True},
}


class 				SCREAMING_SNAKE_CASE__ (__snake_case					):
    __lowerCamelCase						:	Union[str, Any]														=					VOCAB_FILES_NAMES
    __lowerCamelCase						:	int														=					PRETRAINED_VOCAB_FILES_MAP
    __lowerCamelCase						:	Any														=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    __lowerCamelCase						:	int														=					PRETRAINED_INIT_CONFIGURATION
    __lowerCamelCase						:	Tuple														=					RoFormerTokenizer





    def __init__(				self						,       a=None						,       a=None						,       a=True						,       a="[UNK]"						,       a="[SEP]"						,       a="[PAD]"						,       a="[CLS]"						,       a="[MASK]"						,       a=True						,       a=None						,       **a						,       ):
         super().__init__(
             a						,       tokenizer_file=a						,       do_lower_case=a						,       unk_token=a						,       sep_token=a						,       pad_token=a						,       cls_token=a						,       mask_token=a						,       tokenize_chinese_chars=a						,       strip_accents=a						,       **a						,       )

         lowercase__   :       Optional[Any]				=	json.loads(self.backend_tokenizer.normalizer.__getstate__())
         if (
             pre_tok_state.get('lowercase'						,       a) != do_lower_case
             or pre_tok_state.get('strip_accents'						,       a) != strip_accents
         ):
              lowercase__   :       Optional[int]				=	getattr(a						,       pre_tok_state.pop('type'))
              lowercase__   :       str				=	do_lower_case
              lowercase__   :       Union[str, Any]				=	strip_accents
              lowercase__   :       int				=	pre_tok_class(**a)

         lowercase__   :       Optional[int]				=	do_lower_case





    def __getstate__(				self):
         lowercase__   :       str				=	self.__dict__.copy()
         lowercase__   :       Any				=	BertPreTokenizer()
         return state





    def __setstate__(				self						,       a):
         lowercase__   :       Union[str, Any]				=	d
         lowercase__   :       int				=	self.__dict__['_tokenizer'].get_vocab()
         lowercase__   :       List[str]				=	PreTokenizer.custom(JiebaPreTokenizer(a))





    def 				snake_case_    (				self						,       a						,       a=None):
         lowercase__   :       Union[str, Any]				=	[self.cls_token_id] + token_ids_a + [self.sep_token_id]

         if token_ids_a:
              output += token_ids_a + [self.sep_token_id]

         return output





    def 				snake_case_    (				self						,       a						,       a = None):
         lowercase__   :       Tuple				=	[self.sep_token_id]
         lowercase__   :       Optional[int]				=	[self.cls_token_id]
         if token_ids_a is None:
              return len(cls + token_ids_a + sep) * [0]
         return len(cls + token_ids_a + sep) * [0] + len(token_ids_a + sep) * [1]





    def 				snake_case_    (				self						,       a						,       a = None):
         lowercase__   :       Optional[Any]				=	self._tokenizer.model.save(a						,       name=a)
         return tuple(a)





    def 				snake_case_    (				self						,       a						,       a=None						,       a=None						,       a=False						,       **a						,       ):
         lowercase__   :       List[str]				=	BertPreTokenizer()
         return super().save_pretrained(a						,       a						,       a						,       a						,       **a)

