





'''simple docstring'''






import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin




class    _A   (				_a		):
		lowercase__:      Optional[Any]							    =      '''AutoFeatureExtractor'''
		lowercase__:      int							    =      '''Speech2Text2Tokenizer'''

		def __init__(      self				:      Tuple     ,			__magic_name__				:      Tuple     ,			__magic_name__				:      Dict )				->						int:


					"""simple docstring"""

					super().__init__(lowerCamelCase__     ,			lowerCamelCase__ )
					__snake_case				:			Optional[Any]       			=    self.feature_extractor
					__snake_case				:			str       			=    False

		def __call__(      self				:      List[Any]     ,			*__magic_name__				:      Optional[int]     ,			**__magic_name__				:      Optional[Any] )				->						List[Any]:


					"""simple docstring"""

					if self._in_target_context_manager:
								return self.current_processor(*lowerCamelCase__     ,			**lowerCamelCase__ )

					if "raw_speech" in kwargs:
								warnings.warn("""Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead.""" )
								__snake_case				:			List[str]       			=    kwargs.pop("""raw_speech""" )
					else:
								__snake_case				:			Tuple       			=    kwargs.pop("""audio"""     ,			lowerCamelCase__ )
					__snake_case				:			Dict       			=    kwargs.pop("""sampling_rate"""     ,			lowerCamelCase__ )
					__snake_case				:			Dict       			=    kwargs.pop("""text"""     ,			lowerCamelCase__ )
					if len(lowerCamelCase__ ) > 0:
								__snake_case				:			Tuple       			=    args[0]
								__snake_case				:			Optional[int]       			=    args[1:]

					if audio is None and text is None:
								raise ValueError("""You need to specify either an `audio` or `text` input to process.""" )

					if audio is not None:
								__snake_case				:			str       			=    self.feature_extractor(lowerCamelCase__     ,			*lowerCamelCase__     ,			sampling_rate=lowerCamelCase__     ,			**lowerCamelCase__ )
					if text is not None:
								__snake_case				:			Optional[Any]       			=    self.tokenizer(lowerCamelCase__     ,			**lowerCamelCase__ )

					if text is None:
								return inputs
					elif audio is None:
								return encodings
					else:
								__snake_case				:			List[Any]       			=    encodings["input_ids"]
								return inputs

		def   lowercase__    (      self				:      str     ,			*__magic_name__				:      List[str]     ,			**__magic_name__				:      int )				->						str:


					"""simple docstring"""

					return self.tokenizer.batch_decode(*lowerCamelCase__     ,			**lowerCamelCase__ )

		def   lowercase__    (      self				:      Optional[int]     ,			*__magic_name__				:      Union[str, Any]     ,			**__magic_name__				:      int )				->						Tuple:


					"""simple docstring"""

					return self.tokenizer.decode(*lowerCamelCase__     ,			**lowerCamelCase__ )

		@contextmanager
		def   lowercase__    (      self				:      Optional[int] )				->						Tuple:


					"""simple docstring"""

					warnings.warn(
					    """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
					    """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
					    """your audio inputs, or in a separate call.""" )
					__snake_case				:			List[str]       			=    True
					__snake_case				:			str       			=    self.tokenizer
					yield
					__snake_case				:			Any       			=    self.feature_extractor
					__snake_case				:			Optional[int]       			=    False









'''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	(					_lowerCamelCase   )   ->							Optional[Any]:



			"""simple docstring"""


			__snake_case				:			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=2048   ):
												f.write(_lowerCamelCase   )







def 				_a	(					)   ->							VivitConfig:



			"""simple docstring"""


			__snake_case				:			str       			=    VivitConfig()

			__snake_case				:			int       			=    400
			__snake_case				:			str       			=    """huggingface/label-files"""
			__snake_case				:			Any       			=    """kinetics400-id2label.json"""

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







def 				_a	(					)   ->							Any:



			"""simple docstring"""


			__snake_case				:			Any       			=    hf_hub_download(
			    repo_id="""hf-internal-testing/spaghetti-video"""						,  filename="""eating_spaghetti_32_frames.npy"""						,  repo_type="""dataset"""   )
			__snake_case				:			Any       			=    np.load(_lowerCamelCase   )
			return list(_lowerCamelCase   )







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



			"""simple docstring"""


			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 				_a	(					_lowerCamelCase   )   ->							Any:



			"""simple docstring"""


			return current.flatten()







def 				_a	(					_lowerCamelCase   )   ->							Optional[int]:



			"""simple docstring"""


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







def 				_a	(					_lowerCamelCase   )   ->							int:



			"""simple docstring"""


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







def 				_a	(					_lowerCamelCase						,  _lowerCamelCase   )   ->							Dict:



			"""simple docstring"""


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

			__snake_case				:			str       			=    F'''encoder.layer.{i}.'''
			__snake_case				:			Optional[Any]       			=    {
			    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	(					_lowerCamelCase   )   ->							List[Any]:



			"""simple docstring"""


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







def 				_a	(					_lowerCamelCase						,  _lowerCamelCase=False   )   ->							str:



			"""simple docstring"""


			__snake_case				:			Optional[int]       			=    get_n_layers(_lowerCamelCase   )

			__snake_case				:			Optional[Any]       			=    OrderedDict()

			__snake_case				:			Tuple       			=    state_dict["""optimizer"""]["""target"""]["""Transformer"""]["""encoder_norm"""]["""bias"""]
			__snake_case				:			str       			=    state_dict["""optimizer"""]["""target"""]["""Transformer"""]["""encoder_norm"""]["""scale"""]

			__snake_case				:			Optional[Any]       			=    np.transpose(
			    state_dict["""optimizer"""]["""target"""]["""embedding"""]["""kernel"""]						,  (4, 3, 0, 1, 2)   )
			__snake_case				:			Tuple       			=    state_dict["""optimizer"""]["""target"""]["""embedding"""]["""bias"""]

			__snake_case				:			Optional[int]       			=    state_dict["""optimizer"""]["""target"""]["""cls"""]
			__snake_case				:			Dict       			=    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:
						__snake_case				:			Optional[int]       			=    {"""vivit.""" + k: v for k, v in new_state.items()}
						__snake_case				:			Tuple       			=    np.transpose(state_dict["""optimizer"""]["""target"""]["""output_projection"""]["""kernel"""]   )
						__snake_case				:			Tuple       			=    np.transpose(state_dict["""optimizer"""]["""target"""]["""output_projection"""]["""bias"""]   )

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







def 				_a	(					)   ->							VivitImageProcessor:



			"""simple docstring"""


			__snake_case				:			Optional[Any]       			=    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	(					_lowerCamelCase   )   ->							Optional[int]:



			"""simple docstring"""


			__snake_case				:			Optional[int]       			=    """checkpoint"""

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

			__snake_case				:			int       			=    restore_checkpoint(_lowerCamelCase						,  _lowerCamelCase   )
			__snake_case				:			Union[str, Any]       			=    transform_state(_lowerCamelCase						,  classification_head=_lowerCamelCase   )

			__snake_case				:			Optional[Any]       			=    get_vivit_config()

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

			__snake_case				:			List[Any]       			=    VivitForVideoClassification(_lowerCamelCase   )
			model.load_state_dict(_lowerCamelCase   )
			model.eval()

			__snake_case				:			int       			=    get_processor()

			__snake_case				:			int       			=    prepare_video()
			__snake_case				:			List[str]       			=    extractor(_lowerCamelCase						,  return_tensors="""pt"""   )

			__snake_case				:			Optional[int]       			=    model(**_lowerCamelCase   )

			__snake_case				:			List[Any]       			=    torch.Size([1, 400]   )
			__snake_case				:			Tuple       			=    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]						,  _lowerCamelCase						,  atol=1E-4   ), outputs.logits[0, :5]

			model.save_pretrained(_lowerCamelCase   )
			extractor.save_pretrained(_lowerCamelCase   )


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

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

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


