


"""simple docstring"""


import importlib
import importlib.metadata
import os
import warnings
from distutils.util import strtobool
from functools import lru_cache

import torch
from packaging import version
from packaging.version import parse

from .environment import parse_flag_from_env
from .versions import compare_versions, is_torch_version


try:
		import torch_xla.core.xla_model as xm  # noqa: F401

		a_											=     True
except ImportError:
		a_											=     False


# Cache this result has it's a C FFI call which can be pretty time-consuming
a_											=     torch.distributed.is_available()






def   a__			(					__lowercase )						->						Optional[int]:
				# Check we're not importing a "pkg_name" directory somewhere but the actual library by trying to grab the version
				_A       	=      importlib.util.find_spec(__lowercase ) is not None
				if package_exists:
								try:
												_A       	=      importlib.metadata.metadata(__lowercase )
												return True
								except importlib.metadata.PackageNotFoundError:
												return False






def   a__			(					)						->						bool:
				return _torch_distributed_available






def   a__			(					)						->						List[str]:
				try:
								pass
				except ImportError:
								print(
								    "Intel(R) oneCCL Bindings for PyTorch* is required to run DDP on Intel(R) GPUs, but it is not"
								    " detected. If you see \"ValueError: Invalid backend: \'ccl\'\" error, please install Intel(R) oneCCL"
								    " Bindings for PyTorch*." )
				return (
				    importlib.util.find_spec("torch_ccl" ) is not None
				    or importlib.util.find_spec("oneccl_bindings_for_pytorch" ) is not None
				)






def   a__			(					)						->						Optional[int]:
				return importlib.metadata.version("oneccl_bind_pt" )






def   a__			(					)						->						Dict:
				return _is_package_available("transformer_engine" )






@lru_cache
def   a__			(					__lowercase=True )						->						int:
				# Due to bugs on the amp series GPUs, we disable torch-xla on them
				if torch.cuda.is_available():
								return False
				if _tpu_available and check_device:
								try:
												# Will raise a RuntimeError if no XLA configuration is found
												_A       	=      xm.xla_device()
												return True
								except RuntimeError:
												return False
				return _tpu_available






def   a__			(					)						->						Optional[Any]:
				return _is_package_available("deepspeed" )






def   a__			(					__lowercase=False )						->						int:
				if is_tpu_available():
								return not ignore_tpu
				if torch.cuda.is_available():
								return torch.cuda.is_bfaa_supported()
				if is_npu_available():
								return False
				return True






def   a__			(					)						->						Union[str, Any]:
				_A       	=      _is_package_available("bitsandbytes" )
				if package_exists:
								_A       	=      version.parse(importlib.metadata.version("bitsandbytes" ) )
								return compare_versions(__lowercase  ,      ">="  ,      "0.39.0" )
				return False






def   a__			(					)						->						int:
				_A       	=      _is_package_available("bitsandbytes" )
				if package_exists:
								_A       	=      version.parse(importlib.metadata.version("bitsandbytes" ) )
								return compare_versions(__lowercase  ,      ">="  ,      "0.37.2" )
				return False






def   a__			(					)						->						str:
				return _is_package_available("bitsandbytes" )






def   a__			(					)						->						List[str]:
				if strtobool(os.environ.get("ACCELERATE_USE_MEGATRON_LM"  ,      "False" ) ) == 1:
								_A       	=      importlib.util.find_spec("megatron" ) is not None
								if package_exists:
												try:
																_A       	=      parse(importlib.metadata.version("megatron-lm" ) )
																return compare_versions(__lowercase  ,      ">="  ,      "2.2.0" )
												except Exception as e:
																warnings.warn(f"""Parse Megatron version failed. Exception:{e}""" )
																return False






def   a__			(					)						->						Dict:
				return _is_package_available("safetensors" )






def   a__			(					)						->						List[Any]:
				return _is_package_available("transformers" )






def   a__			(					)						->						Any:
				return _is_package_available("datasets" )






def   a__			(					)						->						Union[str, Any]:
				return _is_package_available("aim" )






def   a__			(					)						->						Tuple:
				return _is_package_available("tensorboard" ) or _is_package_available("tensorboardX" )






def   a__			(					)						->						Tuple:
				return _is_package_available("wandb" )






def   a__			(					)						->						List[str]:
				return _is_package_available("comet_ml" )






def   a__			(					)						->						Any:
				return _is_package_available("boto3" )






def   a__			(					)						->						int:
				if _is_package_available("rich" ):
								if "ACCELERATE_DISABLE_RICH" in os.environ:
												warnings.warn(
												    "`ACCELERATE_DISABLE_RICH` is deprecated and will be removed in v0.22.0 and deactivated by default. Please use `ACCELERATE_ENABLE_RICH` if you wish to use `rich`." )
												return not parse_flag_from_env("ACCELERATE_DISABLE_RICH"  ,      __lowercase )
								return parse_flag_from_env("ACCELERATE_ENABLE_RICH"  ,      __lowercase )
				return False






def   a__			(					)						->						int:
				return _is_package_available("sagemaker" )






def   a__			(					)						->						Optional[Any]:
				return _is_package_available("tqdm" )






def   a__			(					)						->						Dict:
				return _is_package_available("mlflow" )






def   a__			(					)						->						Dict:
				return is_torch_version(">="  ,      "1.12" ) and torch.backends.mps.is_available() and torch.backends.mps.is_built()






def   a__			(					)						->						Tuple:
				def get_major_and_minor_from_version(__lowercase ):
								return str(version.parse(__lowercase ).major ) + "." + str(version.parse(__lowercase ).minor )

				_A       	=      importlib.metadata.version("torch" )
				if importlib.util.find_spec("intel_extension_for_pytorch" ) is None:
								return False
				_A       	=      '''N/A'''
				try:
								_A       	=      importlib.metadata.version("intel_extension_for_pytorch" )
				except importlib.metadata.PackageNotFoundError:
								return False
				_A       	=      get_major_and_minor_from_version(_torch_version )
				_A       	=      get_major_and_minor_from_version(_ipex_version )
				if torch_major_and_minor != ipex_major_and_minor:
								warnings.warn(
								    f"""Intel Extension for PyTorch {ipex_major_and_minor} needs to work with PyTorch {ipex_major_and_minor}.*,"""
								    f""" but PyTorch {_torch_version} is found. Please switch to the matching version and run again.""" )
								return False
				return True






@lru_cache
def   a__			(					__lowercase=False )						->						str:
				if importlib.util.find_spec("torch" ) is None or importlib.util.find_spec("torch_npu" ) is None:
								return False

				import torch
				import torch_npu  # noqa: F401

				if check_device:
								try:
												# Will raise a RuntimeError if no NPU is found
												_A       	=      torch.npu.device_count()
												return torch.npu.is_available()
								except RuntimeError:
												return False
				return hasattr(__lowercase  ,      "npu" ) and torch.npu.is_available()






@lru_cache
def   a__			(					__lowercase=False )						->						Any:
				if not parse_flag_from_env("ACCELERATE_USE_XPU"  ,      default=__lowercase ):
								return False
				if is_ipex_available():
								import torch

								if is_torch_version("<="  ,      "1.12" ):
												return False
				else:
								return False

				import intel_extension_for_pytorch  # noqa: F401

				if check_device:
								try:
												# Will raise a RuntimeError if no XPU  is found
												_A       	=      torch.xpu.device_count()
												return torch.xpu.is_available()
								except RuntimeError:
												return False
				return hasattr(__lowercase  ,      "xpu" ) and torch.xpu.is_available()


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				:     Dict     )    ->		Any:
  UpperCAmelCase			:				List[str]										=					'''https://storage.googleapis.com/scenic-bucket/vivit/kinetics_400/vivit_base_16x2_unfactorized/checkpoint'''

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

def      a__	(						)    ->		VivitConfig:
  UpperCAmelCase			:				Any										=					VivitConfig()

  UpperCAmelCase			:				Any										=					400
  UpperCAmelCase			:				int										=					'''huggingface/label-files'''
  UpperCAmelCase			:				int										=					'''kinetics400-id2label.json'''

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

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

def      a__	(						UpperCAmelCase				:     np.ndarray     )    ->		Any:
  if np.ndim(UpperCAmelCase     ) == 2:
    return transform_attention_bias(UpperCAmelCase     )

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

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

def      a__	(						UpperCAmelCase				:     np.ndarray     )    ->		List[Any]:
  return current.flatten()

def      a__	(						UpperCAmelCase				:     np.ndarray     )    ->		Optional[Any]:
  return np.reshape(UpperCAmelCase							,					(current.shape[0], current.shape[1] * current.shape[2])     ).T

def      a__	(						UpperCAmelCase				:     np.ndarray     )    ->		Any:
  return np.reshape(UpperCAmelCase							,					(current.shape[0] * current.shape[1], current.shape[2])     ).T

def      a__	(						UpperCAmelCase				:     Dict							,					UpperCAmelCase				:     Union[str, Any]     )    ->		Dict:
  UpperCAmelCase			:				Any										=					state_dict['''optimizer''']['''target''']['''Transformer'''][f'''encoderblock_{i}''']

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

def      a__	(						UpperCAmelCase				:     Any							,					UpperCAmelCase				:     Dict=False     )    ->		Dict:
  UpperCAmelCase			:				int										=					get_n_layers(UpperCAmelCase     )

  UpperCAmelCase			:				List[str]										=					OrderedDict()

  UpperCAmelCase			:				int										=					state_dict['''optimizer''']['''target''']['''Transformer''']['''encoder_norm''']['''bias''']
  UpperCAmelCase			:				Tuple										=					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			:				Optional[int]										=					state_dict['''optimizer''']['''target''']['''cls''']
  UpperCAmelCase			:				str										=					state_dict['''optimizer''']['''target''']['''Transformer''']['''posembed_input'''][
      '''pos_embedding'''
  ]

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

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

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

def      a__	(						)    ->		VivitImageProcessor:
  UpperCAmelCase			:				Optional[int]										=					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				:     str     )    ->		Tuple:
  UpperCAmelCase			:				Union[str, Any]										=					'''checkpoint'''

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

  UpperCAmelCase			:				Tuple										=					restore_checkpoint(UpperCAmelCase							,					UpperCAmelCase     )
  UpperCAmelCase			:				Any										=					transform_state(UpperCAmelCase							,					classification_head=UpperCAmelCase     )

  UpperCAmelCase			:				Union[str, Any]										=					get_vivit_config()

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

  UpperCAmelCase			:				Optional[int]										=					VivitForVideoClassification(UpperCAmelCase     )
  model.load_state_dict(UpperCAmelCase     )
  model.eval()

  UpperCAmelCase			:				List[Any]										=					get_processor()

  UpperCAmelCase			:				str										=					prepare_video()
  UpperCAmelCase			:				Union[str, Any]										=					extractor(UpperCAmelCase							,					return_tensors='''pt'''     )

  UpperCAmelCase			:				Optional[Any]										=					model(**UpperCAmelCase     )

  UpperCAmelCase			:				str										=					torch.Size([1, 400]     )
  UpperCAmelCase			:				int										=					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]							,					UpperCAmelCase							,					atol=1E-4     ), outputs.logits[0, :5]

  model.save_pretrained(UpperCAmelCase     )
  extractor.save_pretrained(UpperCAmelCase     )


if __name__ == "__main__":
     _lowerCamelCase							:  Optional[Any]     =				argparse.ArgumentParser()

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

     _lowerCamelCase							:  str     =				parser.parse_args()
     convert(args.output_model_name)





