

'''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 						snake_case__    ( lowerCamelCase__				:   Union[str, Any]			)   ->   Optional[Any]:
					A_ :     List[str] =							'''https://storage.googleapis.com/scenic-bucket/vivit/kinetics_400/vivit_base_16x2_unfactorized/checkpoint'''

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




def 						snake_case__    ( )   ->   VivitConfig:
					A_ :     Any =							VivitConfig()

					A_ :     int =							4_0_0
					A_ :     Optional[int] =							'''huggingface/label-files'''
					A_ :     Dict =							'''kinetics400-id2label.json'''

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




def 						snake_case__    ( )   ->   Any:
					A_ :     Any =							hf_hub_download(
					    repo_id='''hf-internal-testing/spaghetti-video'''			,		filename='''eating_spaghetti_32_frames.npy'''			,		repo_type='''dataset'''			)
					A_ :     List[Any] =							np.load(lowerCamelCase__			)
					return list(lowerCamelCase__			)




def 						snake_case__    ( lowerCamelCase__				:   int			)   ->   Any:



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

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

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




def 						snake_case__    ( lowerCamelCase__				:   Any			)   ->   Any:
					return current.flatten()




def 						snake_case__    ( lowerCamelCase__				:   int			)   ->   Union[str, Any]:
					return np.reshape(lowerCamelCase__			,		(current.shape[0], current.shape[1] * current.shape[2])			).T




def 						snake_case__    ( lowerCamelCase__				:   List[str]			)   ->   Optional[int]:
					return np.reshape(lowerCamelCase__			,		(current.shape[0] * current.shape[1], current.shape[2])			).T




def 						snake_case__    ( lowerCamelCase__				:   Union[str, Any]			,		lowerCamelCase__				:   Optional[int]			)   ->   int:
					A_ :     List[str] =							state_dict['''optimizer''']['''target''']['''Transformer'''][f'encoderblock_{i}']

					A_ :     int =							f'encoder.layer.{i}.'
					A_ :     Dict =							{
					    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 						snake_case__    ( lowerCamelCase__				:   Union[str, Any]			)   ->   List[Any]:
					return sum([1 if '''encoderblock_''' in k else 0 for k in state_dict['''optimizer''']['''target''']['''Transformer'''].keys()]			)




def 						snake_case__    ( lowerCamelCase__				:   Union[str, Any]			,		lowerCamelCase__				:   Tuple=False			)   ->   Any:
					A_ :     int =							get_n_layers(lowerCamelCase__			)

					A_ :     List[str] =							OrderedDict()

					A_ :     Optional[Any] =							state_dict['''optimizer''']['''target''']['''Transformer''']['''encoder_norm''']['''bias''']
					A_ :     Optional[Any] =							state_dict['''optimizer''']['''target''']['''Transformer''']['''encoder_norm''']['''scale''']

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

					A_ :     Optional[Any] =							state_dict['''optimizer''']['''target''']['''cls''']
					A_ :     List[str] =							state_dict['''optimizer''']['''target''']['''Transformer''']['''posembed_input'''][
					    '''pos_embedding'''
					]

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

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

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




def 						snake_case__    ( )   ->   VivitImageProcessor:
					A_ :     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 						snake_case__    ( lowerCamelCase__				:   Dict			)   ->   List[Any]:
					A_ :     Union[str, Any] =							'''checkpoint'''

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

					A_ :     List[Any] =							restore_checkpoint(lowerCamelCase__			,		lowerCamelCase__			)
					A_ :     List[Any] =							transform_state(lowerCamelCase__			,		classification_head=lowerCamelCase__			)

					A_ :     Dict =							get_vivit_config()

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

					A_ :     List[str] =							VivitForVideoClassification(lowerCamelCase__			)
					model.load_state_dict(lowerCamelCase__			)
					model.eval()

					A_ :     Tuple =							get_processor()

					A_ :     Optional[Any] =							prepare_video()
					A_ :     Optional[Any] =							extractor(lowerCamelCase__			,		return_tensors='''pt'''			)

					A_ :     List[Any] =							model(**lowerCamelCase__			)

					A_ :     Union[str, Any] =							torch.Size([1, 4_0_0]			)
					A_ :     List[Any] =							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]			,		lowerCamelCase__			,		atol=1e-4			), outputs.logits[0, :5]

					model.save_pretrained(lowerCamelCase__			)
					extractor.save_pretrained(lowerCamelCase__			)


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

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

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









'''simple docstring'''




def 						snake_case__    ( lowerCamelCase__				:   float			)   ->   float:
					return -num if num < 0 else num




def 						snake_case__    ( lowerCamelCase__				:   list[int]			)   ->   int:
					if len(lowerCamelCase__			) == 0:
										raise ValueError('''abs_min() arg is an empty sequence'''			)
					A_ :     Any =							x[0]
					for i in x:
										if abs_val(lowerCamelCase__			) < abs_val(lowerCamelCase__			):
															A_ :     Tuple =							i
					return j




def 						snake_case__    ( lowerCamelCase__				:   list[int]			)   ->   int:
					if len(lowerCamelCase__			) == 0:
										raise ValueError('''abs_max() arg is an empty sequence'''			)
					A_ :     List[str] =							x[0]
					for i in x:
										if abs(lowerCamelCase__			) > abs(lowerCamelCase__			):
															A_ :     Any =							i
					return j




def 						snake_case__    ( lowerCamelCase__				:   list[int]			)   ->   int:
					if len(lowerCamelCase__			) == 0:
										raise ValueError('''abs_max_sort() arg is an empty sequence'''			)
					return sorted(lowerCamelCase__			,		key=lowerCamelCase__			)[-1]




def 						snake_case__    ( )   ->   Optional[Any]:
					assert abs_val(0			) == 0
					assert abs_val(3_4			) == 3_4
					assert abs_val(-1_0_0_0_0_0_0_0_0_0_0_0			) == 1_0_0_0_0_0_0_0_0_0_0_0

					A_ :     Dict =							[-3, -1, 2, -1_1]
					assert abs_max(lowerCamelCase__			) == -1_1
					assert abs_max_sort(lowerCamelCase__			) == -1_1
					assert abs_min(lowerCamelCase__			) == -1


if __name__ == "__main__":
							import doctest

							doctest.testmod()

							test_abs_val()
							print(abs_val(-34))  # --> 34



