

'''simple docstring'''




import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import TimesformerConfig
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,
      TimesformerForVideoClassification,
      TimesformerModel,
  )
  from transformers.models.timesformer.modeling_timesformer import TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
  from transformers import VideoMAEImageProcessor



class 							lowercase_						:

  '''simple docstring'''





  def __init__(  self ,			a_ ,			a_=1_3 ,			a_=1_0 ,			a_=3 ,			a_=2 ,			a_=2 ,			a_=True ,			a_=True ,			a_=3_2 ,			a_=5 ,			a_=4 ,			a_=3_7 ,			a_="gelu" ,			a_=0.1 ,			a_=0.1 ,			a_=1_0 ,			a_=0.02 ,			a_="divided_space_time" ,			a_=None ,			)				->					Optional[int]:


    """simple docstring"""
    UpperCAmelCase    				=						parent
    UpperCAmelCase    				=						batch_size
    UpperCAmelCase    				=						image_size
    UpperCAmelCase    				=						num_channels
    UpperCAmelCase    				=						patch_size
    UpperCAmelCase    				=						num_frames
    UpperCAmelCase    				=						is_training
    UpperCAmelCase    				=						use_labels
    UpperCAmelCase    				=						hidden_size
    UpperCAmelCase    				=						num_hidden_layers
    UpperCAmelCase    				=						num_attention_heads
    UpperCAmelCase    				=						intermediate_size
    UpperCAmelCase    				=						hidden_act
    UpperCAmelCase    				=						hidden_dropout_prob
    UpperCAmelCase    				=						attention_probs_dropout_prob
    UpperCAmelCase    				=						attention_type
    UpperCAmelCase    				=						initializer_range
    UpperCAmelCase    				=						scope
    UpperCAmelCase    				=						num_labels

    # in TimeSformer, the number of spatial tokens equals num_frames * num_patches per frame + 1 CLS token
    UpperCAmelCase    				=						(image_size // patch_size) ** 2
    UpperCAmelCase    				=						(num_frames) * self.num_patches_per_frame + 1
  def     snake_case_      (  self  )				->					Optional[Any]:


    """simple docstring"""
    UpperCAmelCase    				=						floats_tensor(
        [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]  )

    UpperCAmelCase    				=						None
    if self.use_labels:
      UpperCAmelCase    				=						ids_tensor([self.batch_size] ,			self.num_labels  )

    UpperCAmelCase    				=						self.get_config()

    return config, pixel_values, labels
  def     snake_case_      (  self  )				->					int:


    """simple docstring"""
    UpperCAmelCase    				=						TimesformerConfig(
        image_size=self.image_size ,			patch_size=self.patch_size ,			num_channels=self.num_channels ,			num_frames=self.num_frames ,			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 ,			initializer_range=self.initializer_range ,			attention_type=self.attention_type ,			)
    UpperCAmelCase    				=						self.num_labels
    return config
  def     snake_case_      (  self ,			a_ ,			a_ ,			a_  )				->					Optional[int]:


    """simple docstring"""
    UpperCAmelCase    				=						TimesformerModel(config=a_  )
    model.to(a_  )
    model.eval()
    UpperCAmelCase    				=						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_  )				->					Union[str, Any]:


    """simple docstring"""
    UpperCAmelCase    				=						TimesformerForVideoClassification(a_  )
    model.to(a_  )
    model.eval()

    UpperCAmelCase    				=						model(a_  )

    # verify the logits shape
    UpperCAmelCase    				=						torch.Size((self.batch_size, self.num_labels)  )
    self.parent.assertEqual(result.logits.shape ,			a_  )





  def     snake_case_      (  self  )				->					Any:


    """simple docstring"""
    UpperCAmelCase    				=						self.prepare_config_and_inputs()
    UpperCAmelCase				,   UpperCAmelCase				,   UpperCAmelCase    				=						config_and_inputs
    UpperCAmelCase    				=						{'pixel_values': pixel_values}
    return config, inputs_dict




@require_torch
class 							lowercase_						(		a			,							a			,							unittest.TestCase			):

  '''simple docstring'''





  __lowerCAmelCase       :			Optional[int]     	= (TimesformerModel, TimesformerForVideoClassification) if is_torch_available() else ()
  __lowerCAmelCase       :			Any     	= (
      {"feature-extraction": TimesformerModel, "video-classification": TimesformerForVideoClassification}
      if is_torch_available()
      else {}
  )

  __lowerCAmelCase       :			Any     	= False
  __lowerCAmelCase       :			int     	= False
  __lowerCAmelCase       :			Optional[Any]     	= False
  __lowerCAmelCase       :			Dict     	= False
  def     snake_case_      (  self  )				->					List[str]:


    """simple docstring"""
    UpperCAmelCase    				=						TimesformerModelTester(self  )
    UpperCAmelCase    				=						ConfigTester(
        self ,			config_class=a_ ,			has_text_modality=a_ ,			hidden_size=3_7  )
  def     snake_case_      (  self ,			a_ ,			a_ ,			a_=False  )				->					str:


    """simple docstring"""
    UpperCAmelCase    				=						copy.deepcopy(a_  )

    if return_labels:
      if model_class in get_values(a_  ):
        UpperCAmelCase    				=						torch.zeros(
            self.model_tester.batch_size ,			dtype=torch.long ,			device=a_  )

    return inputs_dict
  def     snake_case_      (  self  )				->					List[Any]:


    """simple docstring"""
    self.config_tester.run_common_tests()
  @unittest.skip(reason='TimeSformer does not use inputs_embeds'  )
  def     snake_case_      (  self  )				->					Union[str, Any]:


    """simple docstring"""
    pass
  def     snake_case_      (  self  )				->					Optional[Any]:


    """simple docstring"""
    UpperCAmelCase				,   UpperCAmelCase    				=						self.model_tester.prepare_config_and_inputs_for_common()

    for model_class in self.all_model_classes:
      UpperCAmelCase    				=						model_class(a_  )
      self.assertIsInstance(model.get_input_embeddings() ,			(nn.Module)  )
      UpperCAmelCase    				=						model.get_output_embeddings()
      self.assertTrue(x is None or isinstance(a_ ,			nn.Linear  )  )
  def     snake_case_      (  self  )				->					Optional[int]:


    """simple docstring"""
    UpperCAmelCase				,   UpperCAmelCase    				=						self.model_tester.prepare_config_and_inputs_for_common()

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

      UpperCAmelCase    				=						['pixel_values']
      self.assertListEqual(arg_names[:1] ,			a_  )
  def     snake_case_      (  self  )				->					List[Any]:


    """simple docstring"""
    UpperCAmelCase    				=						self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_model(*a_  )
  def     snake_case_      (  self  )				->					Optional[int]:


    """simple docstring"""
    UpperCAmelCase    				=						self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_for_video_classification(*a_  )
  @slow
  def     snake_case_      (  self  )				->					Optional[int]:


    """simple docstring"""
    for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
      UpperCAmelCase    				=						TimesformerModel.from_pretrained(a_  )
      self.assertIsNotNone(a_  )
  def     snake_case_      (  self  )				->					List[str]:


    """simple docstring"""
    if not self.has_attentions:
      pass

    else:
      UpperCAmelCase				,   UpperCAmelCase    				=						self.model_tester.prepare_config_and_inputs_for_common()
      UpperCAmelCase    				=						True

      for model_class in self.all_model_classes:
        UpperCAmelCase    				=						self.model_tester.seq_length
        UpperCAmelCase    				=						self.model_tester.num_frames

        UpperCAmelCase    				=						True
        UpperCAmelCase    				=						False
        UpperCAmelCase    				=						True
        UpperCAmelCase    				=						model_class(a_  )
        model.to(a_  )
        model.eval()
        with torch.no_grad():
          UpperCAmelCase    				=						model(**self._prepare_for_class(a_ ,			a_  )  )
        UpperCAmelCase    				=						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"]
        UpperCAmelCase    				=						True
        UpperCAmelCase    				=						model_class(a_  )
        model.to(a_  )
        model.eval()
        with torch.no_grad():
          UpperCAmelCase    				=						model(**self._prepare_for_class(a_ ,			a_  )  )
        UpperCAmelCase    				=						outputs.attentions
        self.assertEqual(len(a_  ) ,			self.model_tester.num_hidden_layers  )

        # attentions has shape (batch_size x num_frames) x num_heads x (num_patches per frame + 1) x (num_patches per frame + 1)
        self.assertListEqual(
            list(attentions[0].shape[-3:]  ) ,			[self.model_tester.num_attention_heads, seq_len // num_frames + 1, seq_len // num_frames + 1] ,			)
        UpperCAmelCase    				=						len(a_  )

        # Check attention is always last and order is fine
        UpperCAmelCase    				=						True
        UpperCAmelCase    				=						True
        UpperCAmelCase    				=						model_class(a_  )
        model.to(a_  )
        model.eval()
        with torch.no_grad():
          UpperCAmelCase    				=						model(**self._prepare_for_class(a_ ,			a_  )  )

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

        UpperCAmelCase    				=						outputs.attentions

        self.assertEqual(len(a_  ) ,			self.model_tester.num_hidden_layers  )

        # attentions has shape (batch_size x num_frames) x num_heads x (num_patches per frame + 1) x (num_patches per frame + 1)
        self.assertListEqual(
            list(self_attentions[0].shape[-3:]  ) ,			[self.model_tester.num_attention_heads, seq_len // num_frames + 1, seq_len // num_frames + 1] ,			)





  def     snake_case_      (  self  )				->					Dict:


    """simple docstring"""
    def check_hidden_states_output(a_ ,			a_ ,			a_  ):
      UpperCAmelCase    				=						model_class(a_  )
      model.to(a_  )
      model.eval()

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

      UpperCAmelCase    				=						outputs.hidden_states
      UpperCAmelCase    				=						self.model_tester.num_hidden_layers + 1
      self.assertEqual(len(a_  ) ,			a_  )

      UpperCAmelCase    				=						self.model_tester.seq_length

      self.assertListEqual(
          list(hidden_states[0].shape[-2:]  ) ,			[seq_length, self.model_tester.hidden_size] ,			)

    UpperCAmelCase				,   UpperCAmelCase    				=						self.model_tester.prepare_config_and_inputs_for_common()

    for model_class in self.all_model_classes:
      UpperCAmelCase    				=						True
      check_hidden_states_output(a_ ,			a_ ,			a_  )

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

      check_hidden_states_output(a_ ,			a_ ,			a_  )




def    lowerCamelCase__		(    ):
  UpperCAmelCase    				=						hf_hub_download(
      repo_id='hf-internal-testing/spaghetti-video'							, filename='eating_spaghetti.npy'							, repo_type='dataset'    )
  UpperCAmelCase    				=						np.load(SCREAMING_SNAKE_CASE    )
  return list(SCREAMING_SNAKE_CASE    )



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

  '''simple docstring'''





  @cached_property
  def     snake_case_      (  self  )				->					Any:


    """simple docstring"""
    # 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  )				->					str:


    """simple docstring"""
    UpperCAmelCase    				=						TimesformerForVideoClassification.from_pretrained('facebook/timesformer-base-finetuned-k400'  ).to(
        a_  )

    UpperCAmelCase    				=						self.default_image_processor
    UpperCAmelCase    				=						prepare_video()
    UpperCAmelCase    				=						image_processor(video[:8] ,			return_tensors='pt'  ).to(a_  )

    # forward pass
    with torch.no_grad():
      UpperCAmelCase    				=						model(**a_  )

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

    UpperCAmelCase    				=						torch.tensor([-0.3016, -0.7713, -0.4205]  ).to(a_  )

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





'''simple docstring'''




import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


_a					:			Tuple							=    logging.get_logger(__name__)

_a					:			Optional[int]							=    '▁'

_a					:			str							=    {'vocab_file': 'sentencepiece.bpe.model', 'monolingual_vocab_file': 'dict.txt'}

_a					:			Optional[Any]							=    {
    'vocab_file': {
        'vinai/bartpho-syllable': 'https://huggingface.co/vinai/bartpho-syllable/resolve/main/sentencepiece.bpe.model',
    },
    'monolingual_vocab_file': {
        'vinai/bartpho-syllable': 'https://huggingface.co/vinai/bartpho-syllable/resolve/main/dict.txt',
    },
}

_a					:			Union[str, Any]							=    {'vinai/bartpho-syllable': 1_024}



class 							lowercase_						(		a			):

  '''simple docstring'''





  __lowerCAmelCase       :			Dict     	= VOCAB_FILES_NAMES
  __lowerCAmelCase       :			str     	= PRETRAINED_VOCAB_FILES_MAP
  __lowerCAmelCase       :			Any     	= PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  __lowerCAmelCase       :			int     	= ["input_ids", "attention_mask"]
  def __init__(  self ,			a_ ,			a_ ,			a_="<s>" ,			a_="</s>" ,			a_="</s>" ,			a_="<s>" ,			a_="<unk>" ,			a_="<pad>" ,			a_="<mask>" ,			a_ = None ,			**a_ ,			)				->					None:


    """simple docstring"""
    # Mask token behave like a normal word, i.e. include the space before it
    UpperCAmelCase    				=						AddedToken(a_ ,			lstrip=a_ ,			rstrip=a_  ) if isinstance(a_ ,			a_  ) else mask_token

    UpperCAmelCase    				=						{} if sp_model_kwargs is None else sp_model_kwargs

    super().__init__(
        bos_token=a_ ,			eos_token=a_ ,			unk_token=a_ ,			sep_token=a_ ,			cls_token=a_ ,			pad_token=a_ ,			mask_token=a_ ,			sp_model_kwargs=self.sp_model_kwargs ,			**a_ ,			)

    UpperCAmelCase    				=						vocab_file
    UpperCAmelCase    				=						monolingual_vocab_file
    UpperCAmelCase    				=						spm.SentencePieceProcessor(**self.sp_model_kwargs  )
    self.sp_model.Load(str(a_  )  )

    # Load the reduced vocab

    # Keep order of special tokens for backward compatibility
    UpperCAmelCase    				=						{}
    UpperCAmelCase    				=						0
    for token in [bos_token, pad_token, eos_token, unk_token, sep_token, cls_token]:
      if str(a_  ) not in self.fairseq_tokens_to_ids:
        UpperCAmelCase    				=						cnt
        cnt += 1
    with open(a_ ,			'r' ,			encoding='utf-8'  ) as f:
      for line in f.readlines():
        UpperCAmelCase    				=						line.strip().split()[0]
        UpperCAmelCase    				=						len(self.fairseq_tokens_to_ids  )
    if str(a_  ) not in self.fairseq_tokens_to_ids:
      UpperCAmelCase    				=						len(self.fairseq_tokens_to_ids  )

    UpperCAmelCase    				=						{v: k for k, v in self.fairseq_tokens_to_ids.items()}
  def __getstate__(  self  )				->					Any:


    """simple docstring"""
    UpperCAmelCase    				=						self.__dict__.copy()
    UpperCAmelCase    				=						None
    UpperCAmelCase    				=						self.sp_model.serialized_model_proto()
    return state
  def __setstate__(  self ,			a_  )				->					int:


    """simple docstring"""
    UpperCAmelCase    				=						d

    # for backward compatibility
    if not hasattr(self ,			'sp_model_kwargs'  ):
      UpperCAmelCase    				=						{}

    UpperCAmelCase    				=						spm.SentencePieceProcessor(**self.sp_model_kwargs  )
    self.sp_model.LoadFromSerializedProto(self.sp_model_proto  )
  def     snake_case_      (  self ,			a_ ,			a_ = None  )				->					List[int]:


    """simple docstring"""

    if token_ids_a is None:
      return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
    UpperCAmelCase    				=						[self.cls_token_id]
    UpperCAmelCase    				=						[self.sep_token_id]
    return cls + token_ids_a + sep + sep + token_ids_a + sep
  def     snake_case_      (  self ,			a_ ,			a_ = None ,			a_ = False  )				->					List[int]:


    """simple docstring"""

    if already_has_special_tokens:
      return super().get_special_tokens_mask(
          token_ids_a=a_ ,			token_ids_a=a_ ,			already_has_special_tokens=a_  )

    if token_ids_a is None:
      return [1] + ([0] * len(a_  )) + [1]
    return [1] + ([0] * len(a_  )) + [1, 1] + ([0] * len(a_  )) + [1]
  def     snake_case_      (  self ,			a_ ,			a_ = None  )				->					List[int]:


    """simple docstring"""
    UpperCAmelCase    				=						[self.sep_token_id]
    UpperCAmelCase    				=						[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 + sep + token_ids_a + sep  ) * [0]
  @property
  def     snake_case_      (  self  )				->					Union[str, Any]:


    """simple docstring"""
    return len(self.fairseq_ids_to_tokens  )
  def     snake_case_      (  self  )				->					int:


    """simple docstring"""
    UpperCAmelCase    				=						{self.convert_ids_to_tokens(a_  ): i for i in range(self.vocab_size  )}
    vocab.update(self.added_tokens_encoder  )
    return vocab
  def     snake_case_      (  self ,			a_  )				->					List[str]:


    """simple docstring"""
    return self.sp_model.encode(a_ ,			out_type=a_  )
  def     snake_case_      (  self ,			a_  )				->					Any:


    """simple docstring"""
    if token in self.fairseq_tokens_to_ids:
      return self.fairseq_tokens_to_ids[token]
    else:
      return self.unk_token_id
  def     snake_case_      (  self ,			a_  )				->					Dict:


    """simple docstring"""
    return self.fairseq_ids_to_tokens[index]
  def     snake_case_      (  self ,			a_  )				->					Union[str, Any]:


    """simple docstring"""
    UpperCAmelCase    				=						''.join(a_  ).replace(a_ ,			' '  ).strip()
    return out_string





  def     snake_case_      (  self ,			a_ ,			a_ = None  )				->					Tuple[str]:


    """simple docstring"""
    if not os.path.isdir(a_  ):
      logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''  )
      return
    UpperCAmelCase    				=						os.path.join(
        a_ ,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']  )
    UpperCAmelCase    				=						os.path.join(
        a_ ,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['monolingual_vocab_file'] ,			)

    if os.path.abspath(self.vocab_file  ) != os.path.abspath(a_  ) and os.path.isfile(self.vocab_file  ):
      copyfile(self.vocab_file ,			a_  )
    elif not os.path.isfile(self.vocab_file  ):
      with open(a_ ,			'wb'  ) as fi:
        UpperCAmelCase    				=						self.sp_model.serialized_model_proto()
        fi.write(a_  )

    if os.path.abspath(self.monolingual_vocab_file  ) != os.path.abspath(
        a_  ) and os.path.isfile(self.monolingual_vocab_file  ):
      copyfile(self.monolingual_vocab_file ,			a_  )
    elif not os.path.isfile(self.monolingual_vocab_file  ):
      with open(a_ ,			'w' ,			encoding='utf-8'  ) as fp:
        for token in self.fairseq_tokens_to_ids:
          if token not in self.all_special_tokens:
            fp.write(F'''{str(a_  )} \n'''  )

    return out_vocab_file, out_monolingual_vocab_file


