
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 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE  )   ->      Optional[int]:

							"""simple docstring"""






							_UpperCAmelCase  :			Optional[Any]	       =		"https://storage.googleapis.com/scenic-bucket/vivit/kinetics_400/vivit_base_16x2_unfactorized/checkpoint"

							with open(_SCREAMING_SNAKE_CASE      ,	"wb"  ) as f:
														with requests.get(_SCREAMING_SNAKE_CASE      ,	stream=_SCREAMING_SNAKE_CASE  ) as req:
																					for chunk in req.iter_content(chunk_size=2_0_4_8  ):
																												f.write(_SCREAMING_SNAKE_CASE  )




def 		lowerCAmelCase_     (     )   ->      VivitConfig:

							"""simple docstring"""






							_UpperCAmelCase  :			List[Any]	       =		VivitConfig()

							_UpperCAmelCase  :			Optional[int]	       =		4_0_0
							_UpperCAmelCase  :			int	       =		"huggingface/label-files"
							_UpperCAmelCase  :			Tuple	       =		"kinetics400-id2label.json"

							_UpperCAmelCase  :			Optional[int]	       =		json.load(open(hf_hub_download(_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE      ,	repo_type="dataset"  )      ,	"r"  )  )
							_UpperCAmelCase  :			List[str]	       =		{int(_SCREAMING_SNAKE_CASE  ): v for k, v in idalabel.items()}
							_UpperCAmelCase  :			str	       =		idalabel
							_UpperCAmelCase  :			Optional[int]	       =		{v: k for k, v in idalabel.items()}
							return config




def 		lowerCAmelCase_     (     )   ->      Union[str, Any]:

							"""simple docstring"""






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




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE  )   ->      int:

							"""simple docstring"""






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

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

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




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE  )   ->      Union[str, Any]:

							"""simple docstring"""






							return current.flatten()




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE  )   ->      Tuple:

							"""simple docstring"""






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




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE  )   ->      Union[str, Any]:

							"""simple docstring"""






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




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE  )   ->      Optional[Any]:

							"""simple docstring"""






							_UpperCAmelCase  :			Optional[Any]	       =		state_dict["optimizer"]["target"]["Transformer"][F"""encoderblock_{i}"""]

							_UpperCAmelCase  :			List[str]	       =		F"""encoder.layer.{i}."""
							_UpperCAmelCase  :			List[str]	       =		{
							    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 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE  )   ->      Optional[int]:

							"""simple docstring"""






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




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE=False  )   ->      Optional[int]:

							"""simple docstring"""






							_UpperCAmelCase  :			str	       =		get_n_layers(_SCREAMING_SNAKE_CASE  )

							_UpperCAmelCase  :			List[str]	       =		OrderedDict()

							_UpperCAmelCase  :			Union[str, Any]	       =		state_dict["optimizer"]["target"]["Transformer"]["encoder_norm"]["bias"]
							_UpperCAmelCase  :			Dict	       =		state_dict["optimizer"]["target"]["Transformer"]["encoder_norm"]["scale"]

							_UpperCAmelCase  :			Union[str, Any]	       =		np.transpose(
							    state_dict["optimizer"]["target"]["embedding"]["kernel"]      ,	(4, 3, 0, 1, 2)  )
							_UpperCAmelCase  :			List[str]	       =		state_dict["optimizer"]["target"]["embedding"]["bias"]

							_UpperCAmelCase  :			List[Any]	       =		state_dict["optimizer"]["target"]["cls"]
							_UpperCAmelCase  :			Optional[Any]	       =		state_dict["optimizer"]["target"]["Transformer"]["posembed_input"][
							    "pos_embedding"
							]

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

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

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




def 		lowerCAmelCase_     (     )   ->      VivitImageProcessor:

							"""simple docstring"""






							_UpperCAmelCase  :			str	       =		VivitImageProcessor()

							assert extractor.do_resize is True
							assert extractor.size == {"shortest_edge": 2_5_6}
							assert extractor.do_center_crop is True
							assert extractor.crop_size == {"width": 2_2_4, "height": 2_2_4}
							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 / 2_5_5

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

							return extractor




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE  )   ->      Union[str, Any]:

							"""simple docstring"""






							_UpperCAmelCase  :			str	       =		"checkpoint"

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

							_UpperCAmelCase  :			Union[str, Any]	       =		restore_checkpoint(_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE  )
							_UpperCAmelCase  :			Dict	       =		transform_state(_SCREAMING_SNAKE_CASE      ,	classification_head=_SCREAMING_SNAKE_CASE  )

							_UpperCAmelCase  :			Any	       =		get_vivit_config()

							assert config.image_size == 2_2_4
							assert config.num_frames == 3_2

							_UpperCAmelCase  :			Tuple	       =		VivitForVideoClassification(_SCREAMING_SNAKE_CASE  )
							model.load_state_dict(_SCREAMING_SNAKE_CASE  )
							model.eval()

							_UpperCAmelCase  :			Any	       =		get_processor()

							_UpperCAmelCase  :			Optional[int]	       =		prepare_video()
							_UpperCAmelCase  :			List[Any]	       =		extractor(_SCREAMING_SNAKE_CASE      ,	return_tensors="pt"  )

							_UpperCAmelCase  :			Optional[Any]	       =		model(**_SCREAMING_SNAKE_CASE  )

							_UpperCAmelCase  :			Any	       =		torch.Size([1, 4_0_0]  )
							_UpperCAmelCase  :			List[str]	       =		torch.tensor([-1.0543, 2.0764, -0.2104, 0.4439, -0.9658]  )

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

							model.save_pretrained(_SCREAMING_SNAKE_CASE  )
							extractor.save_pretrained(_SCREAMING_SNAKE_CASE  )


if __name__ == "__main__":
				__lowerCamelCase								=					argparse.ArgumentParser()

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

				__lowerCamelCase								=					parser.parse_args()
				convert(args.output_model_name)






def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE  )   ->      Dict:

							"""simple docstring"""






							if power < 0:
														return -1
							base %= mod
							_UpperCAmelCase  :			Optional[Any]	       =		1

							while power > 0:
														if power & 1:
																					_UpperCAmelCase  :			Optional[Any]	       =		(result * base) % mod
														_UpperCAmelCase  :			List[str]	       =		power >> 1
														_UpperCAmelCase  :			Optional[int]	       =		(base * base) % mod

							return result




def 		lowerCAmelCase_     (     )   ->      List[str]:

							"""simple docstring"""






							print(modular_exponential(3      ,	2_0_0      ,	1_3  )  )


if __name__ == "__main__":
				import doctest

				doctest.testmod()

				main()




