



"""simple docstring"""






import argparse
import json
import os.path
from collections import OrderedDict

import numpy as np
import requests
import torch
from flax.training.checkpoints import restore_checkpoint
from huggingface_hub import hf_hub_download

from transformers import VivitConfig, VivitForVideoClassification, VivitImageProcessor
from transformers.image_utils import PILImageResampling


def 							A__ (     UpperCamelCase__    ):


  '''simple docstring'''


  _SCREAMING_SNAKE_CASE        =			'''https://storage.googleapis.com/scenic-bucket/vivit/kinetics_400/vivit_base_16x2_unfactorized/checkpoint'''

  with open(_A       ,							'''wb'''    ) as f:
    with requests.get(_A       ,							stream=_A    ) as req:
      for chunk in req.iter_content(chunk_size=2_048    ):
        f.write(_A    )


def 							A__ (     ):


  '''simple docstring'''


  _SCREAMING_SNAKE_CASE        =			VivitConfig()

  _SCREAMING_SNAKE_CASE        =			400
  _SCREAMING_SNAKE_CASE        =			'''huggingface/label-files'''
  _SCREAMING_SNAKE_CASE        =			'''kinetics400-id2label.json'''

  _SCREAMING_SNAKE_CASE        =			json.load(open(hf_hub_download(_A       ,							_A       ,							repo_type='''dataset'''    )       ,							'''r'''    )    )
  _SCREAMING_SNAKE_CASE        =			{int(_A    ): v for k, v in idalabel.items()}
  _SCREAMING_SNAKE_CASE        =			idalabel
  _SCREAMING_SNAKE_CASE        =			{v: k for k, v in idalabel.items()}
  return config


def 							A__ (     ):


  '''simple docstring'''


  _SCREAMING_SNAKE_CASE        =			hf_hub_download(
      repo_id='''hf-internal-testing/spaghetti-video'''       ,							filename='''eating_spaghetti_32_frames.npy'''       ,							repo_type='''dataset'''    )
  _SCREAMING_SNAKE_CASE        =			np.load(_A    )
  return list(_A    )


def 							A__ (     UpperCamelCase__    ):


  '''simple docstring'''


  if np.ndim(_A    ) == 2:
    return transform_attention_bias(_A    )

  elif np.ndim(_A    ) == 3:
    return transform_attention_kernel(_A    )

  else:
    raise Exception(F'''Invalid number of dimesions: {np.ndim(_A    )}'''    )


def 							A__ (     UpperCamelCase__    ):


  '''simple docstring'''


  return current.flatten()


def 							A__ (     UpperCamelCase__    ):


  '''simple docstring'''


  return np.reshape(_A       ,							(current.shape[0], current.shape[1] * current.shape[2])    ).T


def 							A__ (     UpperCamelCase__    ):


  '''simple docstring'''


  return np.reshape(_A       ,							(current.shape[0] * current.shape[1], current.shape[2])    ).T


def 							A__ (     UpperCamelCase__       ,							UpperCamelCase__    ):


  '''simple docstring'''


  _SCREAMING_SNAKE_CASE        =			state_dict['''optimizer''']['''target''']['''Transformer'''][F'''encoderblock_{i}''']

  _SCREAMING_SNAKE_CASE        =			F'''encoder.layer.{i}.'''
  _SCREAMING_SNAKE_CASE        =			{
      prefix + '''intermediate.dense.bias''': state['''MlpBlock_0''']['''Dense_0''']['''bias'''],
      prefix + '''intermediate.dense.weight''': np.transpose(state['''MlpBlock_0''']['''Dense_0''']['''kernel''']    ),
      prefix + '''output.dense.bias''': state['''MlpBlock_0''']['''Dense_1''']['''bias'''],
      prefix + '''output.dense.weight''': np.transpose(state['''MlpBlock_0''']['''Dense_1''']['''kernel''']    ),
      prefix + '''layernorm_before.bias''': state['''LayerNorm_0''']['''bias'''],
      prefix + '''layernorm_before.weight''': state['''LayerNorm_0''']['''scale'''],
      prefix + '''layernorm_after.bias''': state['''LayerNorm_1''']['''bias'''],
      prefix + '''layernorm_after.weight''': state['''LayerNorm_1''']['''scale'''],
      prefix
      + '''attention.attention.query.bias''': transform_attention(
          state['''MultiHeadDotProductAttention_0''']['''query''']['''bias''']    ),
      prefix
      + '''attention.attention.query.weight''': transform_attention(
          state['''MultiHeadDotProductAttention_0''']['''query''']['''kernel''']    ),
      prefix
      + '''attention.attention.key.bias''': transform_attention(state['''MultiHeadDotProductAttention_0''']['''key''']['''bias''']    ),
      prefix
      + '''attention.attention.key.weight''': transform_attention(
          state['''MultiHeadDotProductAttention_0''']['''key''']['''kernel''']    ),
      prefix
      + '''attention.attention.value.bias''': transform_attention(
          state['''MultiHeadDotProductAttention_0''']['''value''']['''bias''']    ),
      prefix
      + '''attention.attention.value.weight''': transform_attention(
          state['''MultiHeadDotProductAttention_0''']['''value''']['''kernel''']    ),
      prefix + '''attention.output.dense.bias''': state['''MultiHeadDotProductAttention_0''']['''out''']['''bias'''],
      prefix
      + '''attention.output.dense.weight''': transform_attention_output_weight(
          state['''MultiHeadDotProductAttention_0''']['''out''']['''kernel''']    ),
  }

  return new_state


def 							A__ (     UpperCamelCase__    ):


  '''simple docstring'''


  return sum([1 if '''encoderblock_''' in k else 0 for k in state_dict['''optimizer''']['''target''']['''Transformer'''].keys()]    )


def 							A__ (     UpperCamelCase__       ,							UpperCamelCase__=False    ):


  '''simple docstring'''


  _SCREAMING_SNAKE_CASE        =			get_n_layers(_A    )

  _SCREAMING_SNAKE_CASE        =			OrderedDict()

  _SCREAMING_SNAKE_CASE        =			state_dict['''optimizer''']['''target''']['''Transformer''']['''encoder_norm''']['''bias''']
  _SCREAMING_SNAKE_CASE        =			state_dict['''optimizer''']['''target''']['''Transformer''']['''encoder_norm''']['''scale''']

  _SCREAMING_SNAKE_CASE        =			np.transpose(
      state_dict['''optimizer''']['''target''']['''embedding''']['''kernel''']       ,							(4, 3, 0, 1, 2)    )
  _SCREAMING_SNAKE_CASE        =			state_dict['''optimizer''']['''target''']['''embedding''']['''bias''']

  _SCREAMING_SNAKE_CASE        =			state_dict['''optimizer''']['''target''']['''cls''']
  _SCREAMING_SNAKE_CASE        =			state_dict['''optimizer''']['''target''']['''Transformer''']['''posembed_input'''][
      '''pos_embedding'''
  ]

  for i in range(_A    ):
    new_state.update(transform_state_encoder_block(_A       ,							_A    )    )

  if classification_head:
    _SCREAMING_SNAKE_CASE        =			{'''vivit.''' + k: v for k, v in new_state.items()}
    _SCREAMING_SNAKE_CASE        =			np.transpose(state_dict['''optimizer''']['''target''']['''output_projection''']['''kernel''']    )
    _SCREAMING_SNAKE_CASE        =			np.transpose(state_dict['''optimizer''']['''target''']['''output_projection''']['''bias''']    )

  return {k: torch.tensor(_A    ) for k, v in new_state.items()}


def 							A__ (     ):


  '''simple docstring'''


  _SCREAMING_SNAKE_CASE        =			VivitImageProcessor()

  assert extractor.do_resize is True
  assert extractor.size == {"shortest_edge": 256}
  assert extractor.do_center_crop is True
  assert extractor.crop_size == {"width": 224, "height": 224}
  assert extractor.resample == PILImageResampling.BILINEAR

  # here: https://github.com/deepmind/dmvr/blob/master/dmvr/modalities.py
  # one can seen that add_image has default values for normalization_mean and normalization_std set to 0 and 1
  # which effectively means no normalization (and ViViT does not overwrite those when calling this func)
  assert extractor.do_normalize is False
  assert extractor.do_rescale is True
  assert extractor.rescale_factor == 1 / 255

  # zero-centering = True in original implementation
  assert extractor.do_zero_centering is True

  return extractor


def 							A__ (     UpperCamelCase__    ):


  '''simple docstring'''


  _SCREAMING_SNAKE_CASE        =			'''checkpoint'''

  if not os.path.exists(_A    ):
    download_checkpoint(_A    )

  _SCREAMING_SNAKE_CASE        =			restore_checkpoint(_A       ,							_A    )
  _SCREAMING_SNAKE_CASE        =			transform_state(_A       ,							classification_head=_A    )

  _SCREAMING_SNAKE_CASE        =			get_vivit_config()

  assert config.image_size == 224
  assert config.num_frames == 32

  _SCREAMING_SNAKE_CASE        =			VivitForVideoClassification(_A    )
  model.load_state_dict(_A    )
  model.eval()

  _SCREAMING_SNAKE_CASE        =			get_processor()

  _SCREAMING_SNAKE_CASE        =			prepare_video()
  _SCREAMING_SNAKE_CASE        =			extractor(_A       ,							return_tensors='''pt'''    )

  _SCREAMING_SNAKE_CASE        =			model(**_A    )

  _SCREAMING_SNAKE_CASE        =			torch.Size([1, 400]    )
  _SCREAMING_SNAKE_CASE        =			torch.tensor([-1.05_43, 2.07_64, -0.21_04, 0.44_39, -0.96_58]    )

  assert outputs.logits.shape == expected_shape
  assert torch.allclose(outputs.logits[0, :5]       ,							_A       ,							atol=1e-4    ), outputs.logits[0, :5]

  model.save_pretrained(_A    )
  extractor.save_pretrained(_A    )


if __name__ == "__main__":
    lowerCamelCase				:     Union[str, Any]								=  argparse.ArgumentParser()

    parser.add_argument("""--output_model_name""", """-o""", type=str, help="""Output path for the converted HuggingFace model""")

    lowerCamelCase				:     Dict								=  parser.parse_args()
    convert(args.output_model_name)






from typing import List, Optional, Union

import numpy as np

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


UpperCamelCase__       				=	logging.get_logger(__name__)
class 							__lowercase   (			a__    ):
     _lowerCAmelCase		       =						["input_values", "attention_mask"]



     def __init__(  self   :							Union[str, Any]      ,		lowercase__   :							str=1      ,		lowercase__   :							int=1_6_0_0_0      ,		lowercase__   :							Optional[int]=0.0      ,		lowercase__   :							List[Any]=False      ,		lowercase__   :							Dict=True      ,		**lowercase__   :							Any      ,		):
       super().__init__(feature_size=lowercase__      ,		sampling_rate=lowercase__      ,		padding_value=lowercase__      ,		**lowercase__       )
       a_		    =  return_attention_mask
       a_		    =  do_normalize



     @staticmethod
     def 				__magic_name__		(  lowercase__   :							List[np.ndarray]      ,		lowercase__   :							List[np.ndarray]      ,		lowercase__   :							float = 0.0       ):
       if attention_mask is not None:
         a_		    =  np.array(lowercase__      ,		np.intaa       )
         a_		    =  []

         for vector, length in zip(lowercase__      ,		attention_mask.sum(-1       )       ):
           a_		    =  (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1e-7       )
           if length < normed_slice.shape[0]:
             a_		    =  padding_value

           normed_input_values.append(lowercase__       )
       else:
         a_		    =  [(x - x.mean()) / np.sqrt(x.var() + 1e-7       ) for x in input_values]

       return normed_input_values



     def __call__(  self   :							Any      ,		lowercase__   :							Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]      ,		lowercase__   :							Union[bool, str, PaddingStrategy] = False      ,		lowercase__   :							Optional[int] = None      ,		lowercase__   :							bool = False      ,		lowercase__   :							Optional[int] = None      ,		lowercase__   :							Optional[bool] = None      ,		lowercase__   :							Optional[Union[str, TensorType]] = None      ,		lowercase__   :							Optional[int] = None      ,		**lowercase__   :							List[str]      ,		):

       if sampling_rate is not None:
         if sampling_rate != self.sampling_rate:
           raise ValueError(
               f"The model corresponding to this feature extractor: {self} was trained using a sampling rate of"
               f" {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with"
               f" {self.sampling_rate} and not {sampling_rate}."       )
       else:
         logger.warning(
             '''It is strongly recommended to pass the ``sampling_rate`` argument to this function. '''
             '''Failing to do so can result in silent errors that might be hard to debug.'''       )

       a_		    =  isinstance(lowercase__      ,		np.ndarray       ) and len(raw_speech.shape       ) > 1
       if is_batched_numpy and len(raw_speech.shape       ) > 2:
         raise ValueError(f"Only mono-channel audio is supported for input to {self}"       )
       a_		    =  is_batched_numpy or (
           isinstance(lowercase__      ,		(list, tuple)       ) and (isinstance(raw_speech[0]      ,		(np.ndarray, tuple, list)       ))
       )

       # always return batch
       if not is_batched:
         a_		    =  [raw_speech]

       # convert into correct format for padding
       a_		    =  BatchFeature({'''input_values''': raw_speech}       )

       a_		    =  self.pad(
           lowercase__      ,		padding=lowercase__      ,		max_length=lowercase__      ,		truncation=lowercase__      ,		pad_to_multiple_of=lowercase__      ,		return_attention_mask=lowercase__      ,		)

       # convert input values to correct format
       a_		    =  padded_inputs['''input_values''']
       if not isinstance(input_values[0]      ,		np.ndarray       ):
         a_		    =  [np.asarray(lowercase__      ,		dtype=np.floataa       ) for array in input_values]
       elif (
           not isinstance(lowercase__      ,		np.ndarray       )
           and isinstance(input_values[0]      ,		np.ndarray       )
           and input_values[0].dtype is np.dtype(np.floataa       )
       ):
         a_		    =  [array.astype(np.floataa       ) for array in input_values]
       elif isinstance(lowercase__      ,		np.ndarray       ) and input_values.dtype is np.dtype(np.floataa       ):
         a_		    =  input_values.astype(np.floataa       )

       # convert attention_mask to correct format
       a_		    =  padded_inputs.get('''attention_mask'''       )
       if attention_mask is not None:
         a_		    =  [np.asarray(lowercase__      ,		dtype=np.intaa       ) for array in attention_mask]

       # zero-mean and unit-variance normalization
       if self.do_normalize:
         a_		    =  (
             attention_mask
             if self._get_padding_strategies(lowercase__      ,		max_length=lowercase__       ) is not PaddingStrategy.DO_NOT_PAD
             else None
         )
         a_		    =  self.zero_mean_unit_var_norm(
             padded_inputs['''input_values''']      ,		attention_mask=lowercase__      ,		padding_value=self.padding_value       )

       if return_tensors is not None:
         a_		    =  padded_inputs.convert_to_tensors(lowercase__       )

       return padded_inputs
