





'''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

					snake_case_  :							Union[str, Any]            =	True
except ImportError:
					snake_case_  :							List[str]            =	False


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





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





def 							lowerCamelCase_   (	)			->						bool:
						return _torch_distributed_available





def 							lowerCamelCase_   (	)			->						Any:
						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 							lowerCamelCase_   (	)			->						Tuple:
						return importlib.metadata.version('''oneccl_bind_pt''' )





def 							lowerCamelCase_   (	)			->						int:
						return _is_package_available('''transformer_engine''' )





@lru_cache
def 							lowerCamelCase_   (	SCREAMING_SNAKE_CASE__   :      Optional[int]=True )			->						Dict:
						# 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
																		UpperCAmelCase_      :   List[str]      =	xm.xla_device()
																		return True
												except RuntimeError:
																		return False
						return _tpu_available





def 							lowerCamelCase_   (	)			->						List[Any]:
						return _is_package_available('''deepspeed''' )





def 							lowerCamelCase_   (	SCREAMING_SNAKE_CASE__   :      Optional[Any]=False )			->						Union[str, Any]:
						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 							lowerCamelCase_   (	)			->						Tuple:
						UpperCAmelCase_      :   List[str]      =	_is_package_available('''bitsandbytes''' )
						if package_exists:
												UpperCAmelCase_      :   Optional[int]      =	version.parse(importlib.metadata.version('''bitsandbytes''' ) )
												return compare_versions(SCREAMING_SNAKE_CASE__,      '''>=''',      '''0.39.0''' )
						return False





def 							lowerCamelCase_   (	)			->						Union[str, Any]:
						UpperCAmelCase_      :   List[Any]      =	_is_package_available('''bitsandbytes''' )
						if package_exists:
												UpperCAmelCase_      :   Any      =	version.parse(importlib.metadata.version('''bitsandbytes''' ) )
												return compare_versions(SCREAMING_SNAKE_CASE__,      '''>=''',      '''0.37.2''' )
						return False





def 							lowerCamelCase_   (	)			->						int:
						return _is_package_available('''bitsandbytes''' )





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





def 							lowerCamelCase_   (	)			->						Dict:
						return _is_package_available('''safetensors''' )





def 							lowerCamelCase_   (	)			->						List[str]:
						return _is_package_available('''transformers''' )





def 							lowerCamelCase_   (	)			->						str:
						return _is_package_available('''datasets''' )





def 							lowerCamelCase_   (	)			->						Any:
						return _is_package_available('''aim''' )





def 							lowerCamelCase_   (	)			->						List[Any]:
						return _is_package_available('''tensorboard''' ) or _is_package_available('''tensorboardX''' )





def 							lowerCamelCase_   (	)			->						Tuple:
						return _is_package_available('''wandb''' )





def 							lowerCamelCase_   (	)			->						Any:
						return _is_package_available('''comet_ml''' )





def 							lowerCamelCase_   (	)			->						List[Any]:
						return _is_package_available('''boto3''' )





def 							lowerCamelCase_   (	)			->						Union[str, Any]:
						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''',      SCREAMING_SNAKE_CASE__ )
												return parse_flag_from_env('''ACCELERATE_ENABLE_RICH''',      SCREAMING_SNAKE_CASE__ )
						return False





def 							lowerCamelCase_   (	)			->						Union[str, Any]:
						return _is_package_available('''sagemaker''' )





def 							lowerCamelCase_   (	)			->						List[Any]:
						return _is_package_available('''tqdm''' )





def 							lowerCamelCase_   (	)			->						List[Any]:
						return _is_package_available('''mlflow''' )





def 							lowerCamelCase_   (	)			->						Union[str, Any]:
						return is_torch_version('''>=''',      '''1.12''' ) and torch.backends.mps.is_available() and torch.backends.mps.is_built()





def 							lowerCamelCase_   (	)			->						Union[str, Any]:
						def get_major_and_minor_from_version(SCREAMING_SNAKE_CASE__   :      str ):
												return str(version.parse(SCREAMING_SNAKE_CASE__ ).major ) + "." + str(version.parse(SCREAMING_SNAKE_CASE__ ).minor )

						UpperCAmelCase_      :   List[Any]      =	importlib.metadata.version('''torch''' )
						if importlib.util.find_spec('''intel_extension_for_pytorch''' ) is None:
												return False
						UpperCAmelCase_      :   Dict      =	'''N/A'''
						try:
												UpperCAmelCase_      :   Union[str, Any]      =	importlib.metadata.version('''intel_extension_for_pytorch''' )
						except importlib.metadata.PackageNotFoundError:
												return False
						UpperCAmelCase_      :   List[str]      =	get_major_and_minor_from_version(_torch_version )
						UpperCAmelCase_      :   str      =	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 							lowerCamelCase_   (	SCREAMING_SNAKE_CASE__   :      Dict=False )			->						Any:
						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
																		UpperCAmelCase_      :   List[str]      =	torch.npu.device_count()
																		return torch.npu.is_available()
												except RuntimeError:
																		return False
						return hasattr(SCREAMING_SNAKE_CASE__,      '''npu''' ) and torch.npu.is_available()





@lru_cache
def 							lowerCamelCase_   (	SCREAMING_SNAKE_CASE__   :      List[Any]=False )			->						int:
						if not parse_flag_from_env('''ACCELERATE_USE_XPU''',      default=SCREAMING_SNAKE_CASE__ ):
												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
																		UpperCAmelCase_      :   Optional[int]      =	torch.xpu.device_count()
																		return torch.xpu.is_available()
												except RuntimeError:
																		return False
						return hasattr(SCREAMING_SNAKE_CASE__,      '''xpu''' ) and torch.xpu.is_available()






import argparse

import torch

from transformers import ChineseCLIPConfig, ChineseCLIPModel


def SCREAMING_SNAKE_CASE					(  lowercase_:						List[str]		,    lowercase_:						Dict		,    lowercase_:						List[Any]					):
       _UpperCamelCase    ,   _UpperCamelCase    ,   _UpperCamelCase  :							List[Any]	    =	pt_weights[F"{prefix}.in_proj_weight"].chunk(3		,    dim=0					)
       _UpperCamelCase    ,   _UpperCamelCase    ,   _UpperCamelCase  :							Dict	    =	pt_weights[F"{prefix}.in_proj_bias"].chunk(3		,    dim=0					)

       _UpperCamelCase  :							List[str]	    =	pt_weights[F"{prefix}.out_proj.weight"]
       _UpperCamelCase  :							Any	    =	pt_weights[F"{prefix}.out_proj.bias"]

       _UpperCamelCase  :							List[Any]	    =	q_proj
       _UpperCamelCase  :							Union[str, Any]	    =	q_proj_bias

       _UpperCamelCase  :							str	    =	k_proj
       _UpperCamelCase  :							List[str]	    =	k_proj_bias

       _UpperCamelCase  :							Dict	    =	v_proj
       _UpperCamelCase  :							str	    =	v_proj_bias

       _UpperCamelCase  :							List[str]	    =	out_proj_weights
       _UpperCamelCase  :							Any	    =	out_proj_bias


def SCREAMING_SNAKE_CASE					(  lowercase_:						List[str]		,    lowercase_:						List[Any]		,    lowercase_:						Any					):
       copy_linear(hf_mlp.fca		,    lowercase_		,    F"{prefix}.c_fc"					)
       copy_linear(hf_mlp.fca		,    lowercase_		,    F"{prefix}.c_proj"					)


def SCREAMING_SNAKE_CASE					(  lowercase_:						Optional[Any]		,    lowercase_:						List[str]		,    lowercase_:						Any					):
       _UpperCamelCase  :							Dict	    =	pt_weights[F"{prefix}.weight"].data
       _UpperCamelCase  :							List[Any]	    =	pt_weights[F"{prefix}.bias"].data


def SCREAMING_SNAKE_CASE					(  lowercase_:						Any		,    lowercase_:						Optional[int]		,    lowercase_:						Union[str, Any]					):
       # copy layer norms
       copy_linear(hf_layer.layer_norma		,    lowercase_		,    F"{prefix}.ln_1"					)
       copy_linear(hf_layer.layer_norma		,    lowercase_		,    F"{prefix}.ln_2"					)

       # copy MLP
       copy_mlp(hf_layer.mlp		,    lowercase_		,    F"{prefix}.mlp"					)

       # copy attn
       copy_attn_layer(hf_layer.self_attn		,    lowercase_		,    F"{prefix}.attn"					)


def SCREAMING_SNAKE_CASE					(  lowercase_:						str		,    lowercase_:						str		,    lowercase_:						int					):
       for layer_id, hf_layer in enumerate(lowercase_					):
              copy_layer(lowercase_		,    lowercase_		,    F"{prefix}.{layer_id}"					)


def SCREAMING_SNAKE_CASE					(  lowercase_:						Union[str, Any]		,    lowercase_:						List[Any]					):
       # copy projection
       _UpperCamelCase  :							List[str]	    =	pt_weights['text_projection'].data.T

       # copy text encoder
       for name, param in hf_model.text_model.named_parameters():
              _UpperCamelCase  :							Tuple	    =	pt_weights[F"bert.{name}"].data


def SCREAMING_SNAKE_CASE					(  lowercase_:						str		,    lowercase_:						Any					):
       # copy projection
       _UpperCamelCase  :							List[Any]	    =	pt_weights['visual.proj'].data.T

       # copy layer norms
       copy_linear(hf_model.vision_model.pre_layrnorm		,    lowercase_		,    'visual.ln_pre'					)
       copy_linear(hf_model.vision_model.post_layernorm		,    lowercase_		,    'visual.ln_post'					)

       # copy embeddings
       _UpperCamelCase  :							List[Any]	    =	pt_weights['visual.conv1.weight'].data
       _UpperCamelCase  :							List[Any]	    =	pt_weights['visual.class_embedding'].data
       _UpperCamelCase  :							int	    =	pt_weights['visual.positional_embedding'].data

       # copy encoder
       copy_layers(hf_model.vision_model.encoder.layers		,    lowercase_		,    'visual.transformer.resblocks'					)


@torch.no_grad()
def SCREAMING_SNAKE_CASE					(  lowercase_:						Optional[Any]		,    lowercase_:						Optional[int]		,    lowercase_:						List[Any]=None					):
       assert config_path is not None, "Please specify the ChineseCLIP model config of the corresponding model size."
       _UpperCamelCase  :							Optional[Any]	    =	ChineseCLIPConfig.from_pretrained(lowercase_					)

       _UpperCamelCase  :							List[Any]	    =	ChineseCLIPModel(lowercase_					).eval()

       _UpperCamelCase  :							List[Any]	    =	torch.load(lowercase_		,    map_location='cpu'					)['state_dict']
       _UpperCamelCase  :							Union[str, Any]	    =	{(name[7:] if name.startswith('module.'					) else name): value for name, value in pt_weights.items()}

       copy_text_model_and_projection(lowercase_		,    lowercase_					)
       copy_vision_model_and_projection(lowercase_		,    lowercase_					)
       _UpperCamelCase  :							List[Any]	    =	pt_weights['logit_scale'].data

       hf_model.save_pretrained(lowercase_					)


if __name__ == "__main__":
 UpperCAmelCase      		=					argparse.ArgumentParser()
 parser.add_argument(
     '''--pytorch_dump_folder_path''',
     default=None,
     type=str,
     help='''Path to the output folder storing converted hf PyTorch model.''',
 )
 parser.add_argument(
     '''--checkpoint_path''', default=None, type=str, help='''Path to original github format ChineseCLIP checkpoint.'''
 )
 parser.add_argument(
     '''--config_path''', default=None, required=True, type=str, help='''Path to hf config.json of model to convert.'''
 )
 UpperCAmelCase      		=					parser.parse_args()

 convert_chinese_clip_checkpoint(args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path)
 print('''The conversion is finished!''')



