





import os
import re
import warnings
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
      from .tokenization_ta import TaTokenizer
else:
      _lowerCAmelCase 				=     None


_lowerCAmelCase 				=     logging.get_logger(__name__)

_lowerCAmelCase 				=     {"""vocab_file""": """spiece.model""", """tokenizer_file""": """tokenizer.json"""}

_lowerCAmelCase 				=     {
    """vocab_file""": {
        """t5-small""": """https://huggingface.co/t5-small/resolve/main/spiece.model""",
        """t5-base""": """https://huggingface.co/t5-base/resolve/main/spiece.model""",
        """t5-large""": """https://huggingface.co/t5-large/resolve/main/spiece.model""",
        """t5-3b""": """https://huggingface.co/t5-3b/resolve/main/spiece.model""",
        """t5-11b""": """https://huggingface.co/t5-11b/resolve/main/spiece.model""",
    },
    """tokenizer_file""": {
        """t5-small""": """https://huggingface.co/t5-small/resolve/main/tokenizer.json""",
        """t5-base""": """https://huggingface.co/t5-base/resolve/main/tokenizer.json""",
        """t5-large""": """https://huggingface.co/t5-large/resolve/main/tokenizer.json""",
        """t5-3b""": """https://huggingface.co/t5-3b/resolve/main/tokenizer.json""",
        """t5-11b""": """https://huggingface.co/t5-11b/resolve/main/tokenizer.json""",
    },
}


# TODO(PVP) - this should be removed in Transformers v5
_lowerCAmelCase 				=     {
    """t5-small""": 512,
    """t5-base""": 512,
    """t5-large""": 512,
    """t5-3b""": 512,
    """t5-11b""": 512,
}

class    _UpperCAmelCase							(		_lowerCamelCase     ):
      a						       =							VOCAB_FILES_NAMES
      a						       =							PRETRAINED_VOCAB_FILES_MAP
      a						       =							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      a						       =							['''input_ids''', '''attention_mask''']
      a						       =							TaTokenizer

      a						       =							[]
      def __init__(		self      ,       a__=None      ,       a__=None      ,       a__="</s>"      ,       a__="<unk>"      ,       a__="<pad>"      ,       a__=100      ,       a__=None      ,       **a__      ,       ):
         # Add extra_ids to the special token list
         if extra_ids > 0 and additional_special_tokens is None:
            A_     :				List[str]			      =	[F"""<extra_id_{i}>""" for i in range(a__       )]
         elif extra_ids > 0 and additional_special_tokens is not None:
            # Check that we have the right number of extra special tokens
            A_     :				Any			      =	len(set(filter(lambda a__       : bool("""extra_id_""" in str(a__       )       )      ,       a__       )       )       )
            if extra_tokens != extra_ids:
               raise ValueError(
                   F"""Both extra_ids ({extra_ids}) and additional_special_tokens ({additional_special_tokens}) are"""
                   """ provided to T5Tokenizer. In this case the additional_special_tokens must include the extra_ids"""
                   """ tokens"""       )

         super().__init__(
             a__      ,       tokenizer_file=a__      ,       eos_token=a__      ,       unk_token=a__      ,       pad_token=a__      ,       extra_ids=a__      ,       additional_special_tokens=a__      ,       **a__      ,       )

         A_     :				Optional[int]			      =	vocab_file
         A_     :				List[str]			      =	False if not self.vocab_file else True
         A_     :				int			      =	extra_ids
      @staticmethod
      def 		_lowerCamelCase      (		a__      ,       a__      ,       a__       ):
         if pretrained_model_name_or_path in TaTokenizerFast.max_model_input_sizes:
            A_     :				List[str]			      =	TaTokenizerFast.max_model_input_sizes[pretrained_model_name_or_path]
            if init_max_model_length is not None and init_max_model_length != max_model_length:
               return init_max_model_length
            elif init_max_model_length is None:
               warnings.warn(
                   """This tokenizer was incorrectly instantiated with a model max length of"""
                   F""" {deprecated_max_model_length} which will be corrected in Transformers v5.\nFor now, this"""
                   """ behavior is kept to avoid breaking backwards compatibility when padding/encoding with"""
                   """ `truncation is True`.\n- Be aware that you SHOULD NOT rely on"""
                   F""" {pretrained_model_name_or_path} automatically truncating your input to"""
                   F""" {deprecated_max_model_length} when padding/encoding.\n- If you want to encode/pad to sequences"""
                   F""" longer than {deprecated_max_model_length} you can either instantiate this tokenizer with"""
                   """ `model_max_length` or pass `max_length` when encoding/padding.\n- To avoid this warning, please"""
                   """ instantiate this tokenizer with `model_max_length` set to your preferred value."""      ,       a__      ,       )

         return max_model_length
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__ = None       ):
         if not self.can_save_slow_tokenizer:
            raise ValueError(
                """Your fast tokenizer does not have the necessary information to save the vocabulary for a slow """
                """tokenizer."""       )

         if not os.path.isdir(a__       ):
            logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""       )
            return
         A_     :				List[Any]			      =	os.path.join(
             a__      ,       (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]       )

         if os.path.abspath(self.vocab_file       ) != os.path.abspath(a__       ):
            copyfile(self.vocab_file      ,       a__       )
            logger.info(F"""Copy vocab file to {out_vocab_file}"""       )

         return (out_vocab_file,)
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__ = None       ):
         A_     :				Optional[int]			      =	token_ids_a + [self.eos_token_id]
         if token_ids_a is None:
            return self.prefix_tokens + token_ids_a
         else:
            A_     :				Dict			      =	token_ids_a + [self.eos_token_id]
            return self.prefix_tokens + token_ids_a + token_ids_a
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__ = None       ):
         A_     :				str			      =	[self.eos_token_id]

         if token_ids_a is None:
            return len(token_ids_a + eos       ) * [0]
         return len(token_ids_a + eos + token_ids_a + eos       ) * [0]
      def 		_lowerCamelCase      (		self       ):
         return list(
             set(filter(lambda a__       : bool(re.search(R"""<extra_id_\d+>"""      ,       a__       )       ) is not None      ,       self.additional_special_tokens       )       )       )






      def 		_lowerCamelCase      (		self       ):
         return [self.convert_tokens_to_ids(a__       ) for token in self.get_sentinel_tokens()]













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

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

         # use this variable to define bool_masked_pos
         A_     :				Dict			      =	int(mask_ratio * self.seq_length       )
      def 		_lowerCamelCase      (		self       ):
         A_     :				Any			      =	floats_tensor(
             [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]       )

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

         A_     :				List[Any]			      =	self.get_config()

         return config, pixel_values, labels
      def 		_lowerCamelCase      (		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 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__       ):
         A_     :				Dict			      =	VideoMAEModel(config=a__       )
         model.to(a__       )
         model.eval()
         A_     :				Dict			      =	model(a__       )
         self.parent.assertEqual(result.last_hidden_state.shape      ,       (self.batch_size, self.seq_length, self.hidden_size)       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__       ):
         A_     :				Optional[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
         A_     :				List[Any]			      =	torch.ones((self.num_masks,)       )
         A_     :				Dict			      =	torch.cat([mask, torch.zeros(self.seq_length - mask.size(0       )       )]       )
         A_     :				int			      =	mask.expand(self.batch_size      ,       -1       ).bool()

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






      def 		_lowerCamelCase      (		self       ):
         A_     :				Union[str, Any]			      =	self.prepare_config_and_inputs()
         A_      ,				A_      ,				A_     :				Optional[Any]			      =	config_and_inputs
         A_     :				str			      =	{"""pixel_values""": pixel_values}
         return config, inputs_dict






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

      a						       =							False
      a						       =							False
      a						       =							False
      a						       =							False
      def 		_lowerCamelCase      (		self       ):
         A_     :				int			      =	VideoMAEModelTester(self       )
         A_     :				Any			      =	ConfigTester(self      ,       config_class=a__      ,       has_text_modality=a__      ,       hidden_size=37       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__=False       ):
         A_     :				Optional[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
            A_     :				List[Any]			      =	torch.ones((self.model_tester.num_masks,)       )
            A_     :				List[str]			      =	torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0       )       )]       )
            A_     :				Union[str, Any]			      =	mask.expand(self.model_tester.batch_size      ,       -1       ).bool()
            A_     :				int			      =	bool_masked_pos.to(a__       )

         if return_labels:
            if model_class in [
                *get_values(a__       ),
            ]:
               A_     :				List[str]			      =	torch.zeros(
                   self.model_tester.batch_size      ,       dtype=torch.long      ,       device=a__       )

         return inputs_dict
      def 		_lowerCamelCase      (		self       ):
         self.config_tester.run_common_tests()
      @unittest.skip(reason="""VideoMAE does not use inputs_embeds"""       )
      def 		_lowerCamelCase      (		self       ):
         pass
      def 		_lowerCamelCase      (		self       ):
         A_      ,				A_     :				Optional[Any]			      =	self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
            A_     :				Tuple			      =	model_class(a__       )
            self.assertIsInstance(model.get_input_embeddings()      ,       (nn.Module)       )
            A_     :				Union[str, Any]			      =	model.get_output_embeddings()
            self.assertTrue(x is None or isinstance(a__      ,       nn.Linear       )       )
      def 		_lowerCamelCase      (		self       ):
         A_      ,				A_     :				Optional[int]			      =	self.model_tester.prepare_config_and_inputs_for_common()

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

            A_     :				str			      =	["""pixel_values"""]
            self.assertListEqual(arg_names[:1]      ,       a__       )
      def 		_lowerCamelCase      (		self       ):
         A_     :				List[Any]			      =	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*a__       )
      def 		_lowerCamelCase      (		self       ):
         A_     :				int			      =	self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_for_pretraining(*a__       )
      @slow
      def 		_lowerCamelCase      (		self       ):
         for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
            A_     :				Any			      =	VideoMAEModel.from_pretrained(a__       )
            self.assertIsNotNone(a__       )
      def 		_lowerCamelCase      (		self       ):
         if not self.has_attentions:
            pass

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

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

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

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

               A_     :				Optional[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 		_lowerCamelCase      (		self       ):
         def check_hidden_states_output(a__      ,       a__      ,       a__       ):
            A_     :				str			      =	model_class(a__       )
            model.to(a__       )
            model.eval()

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

            A_     :				Optional[Any]			      =	outputs.hidden_states
            A_     :				Union[str, Any]			      =	self.model_tester.num_hidden_layers + 1
            self.assertEqual(len(a__       )      ,       a__       )

            A_     :				str			      =	self.model_tester.seq_length - self.model_tester.num_masks
            A_     :				str			      =	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_     :				Any			      =	self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
            A_     :				Union[str, Any]			      =	True
            check_hidden_states_output(a__      ,       a__      ,       a__       )

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






def     _lowerCAmelCase		(  ):
   '''simple docstring'''


   A_     :				int			      =	hf_hub_download(
       repo_id="""hf-internal-testing/spaghetti-video"""  ,filename="""eating_spaghetti.npy"""  ,repo_type="""dataset"""     )
   A_     :				Optional[int]			      =	np.load(_lowerCAmelCase     )
   return list(_lowerCAmelCase     )







@require_torch
@require_vision
class    _UpperCAmelCase							(		unittest.TestCase     ):
      @cached_property
      def 		_lowerCamelCase      (		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 		_lowerCamelCase      (		self       ):
         A_     :				List[str]			      =	VideoMAEForVideoClassification.from_pretrained("""MCG-NJU/videomae-base-finetuned-kinetics"""       ).to(
             a__       )

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

         # forward pass
         with torch.no_grad():
            A_     :				Any			      =	model(**a__       )

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

         A_     :				Optional[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 		_lowerCamelCase      (		self       ):
         A_     :				int			      =	VideoMAEForPreTraining.from_pretrained("""MCG-NJU/videomae-base-short"""       ).to(a__       )

         A_     :				str			      =	self.default_image_processor
         A_     :				Any			      =	prepare_video()
         A_     :				Union[str, Any]			      =	image_processor(a__      ,       return_tensors="""pt"""       ).to(a__       )

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

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

         # verify the logits
         A_     :				int			      =	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=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`)
         A_     :				Optional[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`)
         A_     :				Optional[int]			      =	VideoMAEForPreTraining.from_pretrained("""MCG-NJU/videomae-base-short"""      ,       norm_pix_loss=a__       ).to(
             a__       )

         with torch.no_grad():
            A_     :				Optional[Any]			      =	model(**a__       )

         A_     :				List[Any]			      =	torch.tensor(torch.tensor([0.6469]       )      ,       device=a__       )
         self.assertTrue(torch.allclose(outputs.loss      ,       a__      ,       atol=1E-4       )       )






