




import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import VivitConfig
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


if is_torch_available():
   import torch
   from torch import nn

   from transformers import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING, VivitForVideoClassification, VivitModel
   from transformers.models.vivit.modeling_vivit import VIVIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
   from transformers import VivitImageProcessor

class   UpperCAmelCase    :






   def __init__(						self :							Union[str, Any]    ,       UpperCamelCase_ :							Optional[int]    ,       UpperCamelCase_ :							List[str]=2    ,       UpperCamelCase_ :							Optional[Any]=True    ,       UpperCamelCase_ :							Union[str, Any]=True    ,       UpperCamelCase_ :							List[Any]=10    ,       UpperCamelCase_ :							Optional[Any]=10    ,       UpperCamelCase_ :							Any=8    ,       UpperCamelCase_ :							Optional[int]=[2, 4, 4]    ,       UpperCamelCase_ :							List[str]=3    ,       UpperCamelCase_ :							Tuple=7_68    ,       UpperCamelCase_ :							Optional[int]=5    ,       UpperCamelCase_ :							Optional[Any]=4    ,       UpperCamelCase_ :							Union[str, Any]=37    ,       UpperCamelCase_ :							List[Any]="gelu_fast"    ,       UpperCamelCase_ :							int=0.0    ,       UpperCamelCase_ :							Any=0.0    ,       UpperCamelCase_ :							int=0.02    ,       UpperCamelCase_ :							Optional[Any]=1E-06    ,       UpperCamelCase_ :							str=True    ,       UpperCamelCase_ :							Optional[Any]=None    ,       )->      Dict:
     _lowercase:			List[str]      =     parent
     _lowercase:			List[str]      =     batch_size
     _lowercase:			Dict      =     is_training
     _lowercase:			Tuple      =     use_labels
     _lowercase:			Union[str, Any]      =     num_labels
     _lowercase:			List[str]      =     image_size
     _lowercase:			List[Any]      =     num_frames
     _lowercase:			List[Any]      =     tubelet_size
     _lowercase:			List[str]      =     num_channels
     _lowercase:			Any      =     hidden_size
     _lowercase:			Dict      =     num_hidden_layers
     _lowercase:			Any      =     num_attention_heads
     _lowercase:			List[str]      =     intermediate_size
     _lowercase:			List[str]      =     hidden_act
     _lowercase:			List[Any]      =     hidden_dropout_prob
     _lowercase:			Union[str, Any]      =     attention_probs_dropout_prob
     _lowercase:			Union[str, Any]      =     initializer_range
     _lowercase:			List[str]      =     layer_norm_eps
     _lowercase:			str      =     qkv_bias
     _lowercase:			int      =     scope

     _lowercase:			Tuple      =     (
         (self.image_size // self.tubelet_size[2])
         * (self.image_size // self.tubelet_size[1])
         * (self.num_frames // self.tubelet_size[0])
     ) + 1  # CLS token






   def 			_a    (						self :							List[Any]  )->      Tuple:
     _lowercase:			Optional[int]      =     floats_tensor(
         [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]  )

     _lowercase:			Union[str, Any]      =     None
     if self.use_labels:
       _lowercase:			Any      =     ids_tensor([self.batch_size]    ,       self.num_labels  )

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

     return config, pixel_values, labels






   def 			_a    (						self :							List[str]  )->      int:
     _lowercase:			Union[str, Any]      =     VivitConfig(
         num_frames=self.num_frames    ,       image_size=self.image_size    ,       tubelet_size=self.tubelet_size    ,       num_channels=self.num_channels    ,       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    ,       layer_norm_eps=self.layer_norm_eps    ,       qkv_bias=self.qkv_bias    ,       )
     _lowercase:			Optional[Any]      =     self.num_labels
     return config






   def 			_a    (						self :							Optional[int]    ,       UpperCamelCase_ :							Optional[int]    ,       UpperCamelCase_ :							Any    ,       UpperCamelCase_ :							str  )->      int:
     _lowercase:			List[str]      =     VivitModel(config=UpperCamelCase_  )
     model.to(UpperCamelCase_  )
     model.eval()
     _lowercase:			List[Any]      =     model(UpperCamelCase_  )
     self.parent.assertEqual(result.last_hidden_state.shape    ,       (self.batch_size, self.seq_length, self.hidden_size)  )






   def 			_a    (						self :							Any    ,       UpperCamelCase_ :							List[str]    ,       UpperCamelCase_ :							List[str]    ,       UpperCamelCase_ :							Any  )->      Optional[int]:
     _lowercase:			Optional[int]      =     VivitForVideoClassification(UpperCamelCase_  )
     model.to(UpperCamelCase_  )
     model.eval()

     _lowercase:			Any      =     model(UpperCamelCase_  )

     # verify the logits shape
     _lowercase:			int      =     torch.Size((self.batch_size, self.num_labels)  )
     self.parent.assertEqual(result.logits.shape    ,       UpperCamelCase_  )






   def 			_a    (						self :							Union[str, Any]  )->      Union[str, Any]:
     _lowercase:			int      =     self.prepare_config_and_inputs()
     _lowercase  ,       _lowercase  ,       _lowercase:			Any      =     config_and_inputs
     _lowercase:			List[str]      =     {'pixel_values': pixel_values}
     return config, inputs_dict

@require_torch
class   UpperCAmelCase    (      A ,       unittest.TestCase ):
   __SCREAMING_SNAKE_CASE		:				Tuple      					=       (VivitModel, VivitForVideoClassification) if is_torch_available() else ()

   __SCREAMING_SNAKE_CASE		:				Optional[Any]      					=       False
   __SCREAMING_SNAKE_CASE		:				List[Any]      					=       False
   __SCREAMING_SNAKE_CASE		:				int      					=       False
   __SCREAMING_SNAKE_CASE		:				str      					=       False






   def 			_a    (						self :							Dict  )->      Optional[int]:
     _lowercase:			Union[str, Any]      =     VivitModelTester(self  )
     _lowercase:			List[str]      =     ConfigTester(self    ,       config_class=UpperCamelCase_    ,       has_text_modality=UpperCamelCase_    ,       hidden_size=37  )






   def 			_a    (						self :							Union[str, Any]    ,       UpperCamelCase_ :							Optional[int]    ,       UpperCamelCase_ :							List[Any]    ,       UpperCamelCase_ :							List[Any]=False  )->      str:
     _lowercase:			List[str]      =     copy.deepcopy(UpperCamelCase_  )

     if return_labels:
       if model_class in get_values(UpperCamelCase_  ):
         _lowercase:			Union[str, Any]      =     torch.zeros(
             self.model_tester.batch_size    ,       dtype=torch.long    ,       device=UpperCamelCase_  )

     return inputs_dict






   def 			_a    (						self :							int  )->      List[Any]:
     self.config_tester.run_common_tests()






   @unittest.skip(reason='Vivit does not use inputs_embeds'  )
   def 			_a    (						self :							Dict  )->      Dict:
     pass






   def 			_a    (						self :							Any  )->      Union[str, Any]:
     _lowercase  ,       _lowercase:			Optional[int]      =     self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
       _lowercase:			List[Any]      =     model_class(UpperCamelCase_  )
       self.assertIsInstance(model.get_input_embeddings()    ,       (nn.Module)  )
       _lowercase:			Tuple      =     model.get_output_embeddings()
       self.assertTrue(x is None or isinstance(UpperCamelCase_    ,       nn.Linear  )  )






   def 			_a    (						self :							int  )->      Any:
     _lowercase  ,       _lowercase:			List[str]      =     self.model_tester.prepare_config_and_inputs_for_common()

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

       _lowercase:			Any      =     ['pixel_values', 'head_mask']
       self.assertListEqual(arg_names[:2]    ,       UpperCamelCase_  )






   def 			_a    (						self :							Optional[Any]  )->      Any:
     _lowercase:			int      =     self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_model(*UpperCamelCase_  )






   def 			_a    (						self :							Tuple  )->      Union[str, Any]:
     _lowercase:			Any      =     self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_video_classification(*UpperCamelCase_  )






   @slow
   def 			_a    (						self :							Union[str, Any]  )->      Any:
     for model_name in VIVIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
       _lowercase:			List[Any]      =     VivitModel.from_pretrained(UpperCamelCase_  )
       self.assertIsNotNone(UpperCamelCase_  )






   def 			_a    (						self :							List[Any]  )->      str:
     _lowercase  ,       _lowercase:			int      =     self.model_tester.prepare_config_and_inputs_for_common()
     _lowercase:			Any      =     True

     for model_class in self.all_model_classes:
       _lowercase:			Any      =     self.model_tester.seq_length

       _lowercase:			Optional[int]      =     True
       _lowercase:			Any      =     False
       _lowercase:			List[Any]      =     True
       _lowercase:			Optional[int]      =     model_class(UpperCamelCase_  )
       model.to(UpperCamelCase_  )
       model.eval()
       with torch.no_grad():
         _lowercase:			List[Any]      =     model(**self._prepare_for_class(UpperCamelCase_    ,       UpperCamelCase_  )  )
       _lowercase:			Optional[int]      =     outputs.attentions
       self.assertEqual(len(UpperCamelCase_  )    ,       self.model_tester.num_hidden_layers  )

       # check that output_attentions also work using config
       del inputs_dict["output_attentions"]
       _lowercase:			List[Any]      =     True
       _lowercase:			int      =     model_class(UpperCamelCase_  )
       model.to(UpperCamelCase_  )
       model.eval()
       with torch.no_grad():
         _lowercase:			int      =     model(**self._prepare_for_class(UpperCamelCase_    ,       UpperCamelCase_  )  )
       _lowercase:			str      =     outputs.attentions
       self.assertEqual(len(UpperCamelCase_  )    ,       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(UpperCamelCase_  )

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

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

       _lowercase:			Any      =     outputs.attentions

       self.assertEqual(len(UpperCamelCase_  )    ,       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 :							List[str]  )->      Optional[int]:
     def check_hidden_states_output(UpperCamelCase_ :							str    ,       UpperCamelCase_ :							List[str]    ,       UpperCamelCase_ :							Optional[int]  ):
       _lowercase:			List[str]      =     model_class(UpperCamelCase_  )
       model.to(UpperCamelCase_  )
       model.eval()

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

       _lowercase:			List[str]      =     outputs.hidden_states
       _lowercase:			Tuple      =     self.model_tester.num_hidden_layers + 1
       self.assertEqual(len(UpperCamelCase_  )    ,       UpperCamelCase_  )

       _lowercase:			List[Any]      =     self.model_tester.seq_length

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

     _lowercase  ,       _lowercase:			str      =     self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
       _lowercase:			Any      =     True
       check_hidden_states_output(UpperCamelCase_    ,       UpperCamelCase_    ,       UpperCamelCase_  )

       # check that output_hidden_states also work using config
       del inputs_dict["output_hidden_states"]
       _lowercase:			Any      =     True

       check_hidden_states_output(UpperCamelCase_    ,       UpperCamelCase_    ,       UpperCamelCase_  )






   @unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.'  )
   def 			_a    (						self :							Union[str, Any]  )->      List[Any]:
     pass


def 		_lowerCamelCase					(       )      ->		List[str]:





  """simple docstring"""





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





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






   @cached_property
   def 			_a    (						self :							Dict  )->      int:
     return VivitImageProcessor() if is_vision_available() else None






   @slow
   def 			_a    (						self :							Any  )->      str:
     _lowercase:			int      =     VivitForVideoClassification.from_pretrained('google/vivit-b-16x2-kinetics400'  ).to(UpperCamelCase_  )

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

     # forward pass
     with torch.no_grad():
       _lowercase:			Any      =     model(**UpperCamelCase_  )

     # verify the logits
     _lowercase:			str      =     torch.Size((1, 4_00)  )
     self.assertEqual(outputs.logits.shape    ,       UpperCamelCase_  )

     # taken from original model
     _lowercase:			str      =     torch.tensor([-1.0_543, 2.0_764, -0.2_104, 0.4_439, -0.9_658]  ).to(UpperCamelCase_  )

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






import os
import sys

from . import rsa_key_generator as rkg

UpperCAmelCase__    :							str       		=     128
UpperCAmelCase__    :							int       		=     256


def 		_lowerCamelCase					(       UpperCamelCase       ,  UpperCamelCase = DEFAULT_BLOCK_SIZE				)      ->		list[int]:





  """simple docstring"""





  _lowercase:			List[str]      =     message.encode('ascii'				)

  _lowercase:			str      =     []
  for block_start in range(0       ,  len(UpperCamelCase				)       ,  UpperCamelCase				):
    _lowercase:			Optional[int]      =     0
    for i in range(UpperCamelCase       ,  min(block_start + block_size       ,  len(UpperCamelCase				)				)				):
      block_int += message_bytes[i] * (BYTE_SIZE ** (i % block_size))
    block_ints.append(UpperCamelCase				)
  return block_ints


def 		_lowerCamelCase					(       UpperCamelCase       ,  UpperCamelCase       ,  UpperCamelCase = DEFAULT_BLOCK_SIZE				)      ->		str:





  """simple docstring"""





  _lowercase:			list[str]      =     []
  for block_int in block_ints:
    _lowercase:			list[str]      =     []
    for i in range(block_size - 1       ,  -1       ,  -1				):
      if len(UpperCamelCase				) + i < message_length:
        _lowercase:			Optional[Any]      =     block_int // (BYTE_SIZE**i)
        _lowercase:			Dict      =     block_int % (BYTE_SIZE**i)
        block_message.insert(0       ,  chr(UpperCamelCase				)				)
    message.extend(UpperCamelCase				)
  return "".join(UpperCamelCase				)


def 		_lowerCamelCase					(       UpperCamelCase       ,  UpperCamelCase       ,  UpperCamelCase = DEFAULT_BLOCK_SIZE				)      ->		list[int]:





  """simple docstring"""





  _lowercase:			Optional[int]      =     []
  _lowercase  ,       _lowercase:			str      =     key

  for block in get_blocks_from_text(UpperCamelCase       ,  UpperCamelCase				):
    encrypted_blocks.append(pow(UpperCamelCase       ,  UpperCamelCase       ,  UpperCamelCase				)				)
  return encrypted_blocks


def 		_lowerCamelCase					(       UpperCamelCase       ,  UpperCamelCase       ,  UpperCamelCase       ,  UpperCamelCase = DEFAULT_BLOCK_SIZE       ,  )      ->		str:





  """simple docstring"""





  _lowercase:			Optional[int]      =     []
  _lowercase  ,       _lowercase:			str      =     key
  for block in encrypted_blocks:
    decrypted_blocks.append(pow(UpperCamelCase       ,  UpperCamelCase       ,  UpperCamelCase				)				)
  return get_text_from_blocks(UpperCamelCase       ,  UpperCamelCase       ,  UpperCamelCase				)


def 		_lowerCamelCase					(       UpperCamelCase				)      ->		tuple[int, int, int]:





  """simple docstring"""





  with open(UpperCamelCase				) as fo:
    _lowercase:			Any      =     fo.read()
  _lowercase  ,       _lowercase  ,       _lowercase:			Optional[Any]      =     content.split(','				)
  return (int(UpperCamelCase				), int(UpperCamelCase				), int(UpperCamelCase				))


def 		_lowerCamelCase					(       UpperCamelCase       ,  UpperCamelCase       ,  UpperCamelCase       ,  UpperCamelCase = DEFAULT_BLOCK_SIZE       ,  )      ->		str:





  """simple docstring"""





  _lowercase  ,       _lowercase  ,       _lowercase:			int      =     read_key_file(UpperCamelCase				)
  if key_size < block_size * 8:
    sys.exit(
        'ERROR: Block size is {} bits and key size is {} bits. The RSA cipher '
        'requires the block size to be equal to or greater than the key size. '
        'Either decrease the block size or use different keys.'.format(
            block_size * 8       ,  UpperCamelCase				)				)

  _lowercase:			int      =     [str(UpperCamelCase				) for i in encrypt_message(UpperCamelCase       ,  (n, e)       ,  UpperCamelCase				)]

  _lowercase:			List[Any]      =     ','.join(UpperCamelCase				)
  _lowercase:			Tuple      =     F'{len(UpperCamelCase				)}_{block_size}_{encrypted_content}'
  with open(UpperCamelCase       ,  'w'				) as fo:
    fo.write(UpperCamelCase				)
  return encrypted_content


def 		_lowerCamelCase					(       UpperCamelCase       ,  UpperCamelCase				)      ->		str:





  """simple docstring"""





  _lowercase  ,       _lowercase  ,       _lowercase:			Tuple      =     read_key_file(UpperCamelCase				)
  with open(UpperCamelCase				) as fo:
    _lowercase:			Optional[Any]      =     fo.read()
  _lowercase  ,       _lowercase  ,       _lowercase:			Union[str, Any]      =     content.split('_'				)
  _lowercase:			List[Any]      =     int(UpperCamelCase				)
  _lowercase:			Dict      =     int(UpperCamelCase				)

  if key_size < block_size * 8:
    sys.exit(
        'ERROR: Block size is {} bits and key size is {} bits. The RSA cipher '
        'requires the block size to be equal to or greater than the key size. '
        'Did you specify the correct key file and encrypted file?'.format(
            block_size * 8       ,  UpperCamelCase				)				)

  _lowercase:			Union[str, Any]      =     []
  for block in encrypted_message.split(','				):
    encrypted_blocks.append(int(UpperCamelCase				)				)

  return decrypt_message(UpperCamelCase       ,  UpperCamelCase       ,  (n, d)       ,  UpperCamelCase				)


def 		_lowerCamelCase					(       )      ->		None:





  """simple docstring"""





  _lowercase:			Tuple      =     'encrypted_file.txt'
  _lowercase:			Optional[Any]      =     input(r'Encrypt\Decrypt [e\d]: '				)

  if response.lower().startswith('e'				):
    _lowercase:			Optional[Any]      =     'encrypt'
  elif response.lower().startswith('d'				):
    _lowercase:			Union[str, Any]      =     'decrypt'

  if mode == "encrypt":
    if not os.path.exists('rsa_pubkey.txt'				):
      rkg.make_key_files('rsa'       ,  1_024				)

    _lowercase:			Union[str, Any]      =     input('\nEnter message: '				)
    _lowercase:			Tuple      =     'rsa_pubkey.txt'
    print(F'Encrypting and writing to {filename}...'				)
    _lowercase:			Optional[Any]      =     encrypt_and_write_to_file(UpperCamelCase       ,  UpperCamelCase       ,  UpperCamelCase				)

    print('\nEncrypted text:'				)
    print(UpperCamelCase				)

  elif mode == "decrypt":
    _lowercase:			Optional[Any]      =     'rsa_privkey.txt'
    print(F'Reading from {filename} and decrypting...'				)
    _lowercase:			int      =     read_from_file_and_decrypt(UpperCamelCase       ,  UpperCamelCase				)
    print('writing decryption to rsa_decryption.txt...'				)
    with open('rsa_decryption.txt'       ,  'w'				) as dec:
      dec.write(UpperCamelCase				)

    print('\nDecryption:'				)
    print(UpperCamelCase				)


if __name__ == "__main__":
   main()
