





from typing import List, Optional, Tuple, Union

from . import logging
from .import_utils import is_torch_available, is_torch_version


if is_torch_available():
 import torch

_lowerCAmelCase							:      List[str]						=	logging.get_logger(__name__)  # pylint: disable=invalid-name

try:
 from torch._dynamo import allow_in_graph as maybe_allow_in_graph
except (ImportError, ModuleNotFoundError):

 def      __snake_case						(					cls : Optional[Any]	) ->						Union[str, Any]:

    '''simple docstring'''





    return cls

def      __snake_case						(					_lowerCAmelCase : Tuple				,				_lowerCAmelCase : Dict = None				,				_lowerCAmelCase : Optional[int] = None				,				_lowerCAmelCase : int = None				,				_lowerCAmelCase : str = None				,				) ->						List[Any]:

   '''simple docstring'''





   A_					:				List[str]          =			device
   A_					:				Tuple          =			shape[0]

   A_					:				int          =			layout or torch.strided
   A_					:				Any          =			device or torch.device("cpu"	)

   if generator is not None:
      A_					:				Dict          =			generator.device.type if not isinstance(lowercase_				,				lowercase_	) else generator[0].device.type
      if gen_device_type != device.type and gen_device_type == "cpu":
         A_					:				Dict          =			"cpu"
         if device != "mps":
            logger.info(
                f"The passed generator was created on \'cpu\' even though a tensor on {device} was expected."
                f" Tensors will be created on \'cpu\' and then moved to {device}. Note that one can probably"
                f" slighly speed up this function by passing a generator that was created on the {device} device."	)
      elif gen_device_type != device.type and gen_device_type == "cuda":
         raise ValueError(f"Cannot generate a {device} tensor from a generator of type {gen_device_type}."	)

   if isinstance(lowercase_				,				lowercase_	):
      A_					:				Any          =			(1,) + shape[1:]
      A_					:				int          =			[
          torch.randn(lowercase_				,				generator=generator[i]				,				device=lowercase_				,				dtype=lowercase_				,				layout=lowercase_	)
          for i in range(lowercase_	)
      ]
      A_					:				int          =			torch.cat(lowercase_				,				dim=0	).to(lowercase_	)
   else:
      A_					:				List[Any]          =			torch.randn(lowercase_				,				generator=lowercase_				,				device=lowercase_				,				dtype=lowercase_				,				layout=lowercase_	).to(lowercase_	)

   return latents





def      __snake_case						(					_lowerCAmelCase : Optional[int]	) ->						List[str]:

   '''simple docstring'''





   if is_torch_version("<"				,				"2.0.0"	) or not hasattr(lowercase_				,				"_dynamo"	):
      return False
   return isinstance(lowercase_				,				torch._dynamo.eval_frame.OptimizedModule	)







import argparse

import torch

from transformers import ChineseCLIPConfig, ChineseCLIPModel

def      __snake_case						(					_lowerCAmelCase : str				,				_lowerCAmelCase : Optional[int]				,				_lowerCAmelCase : Optional[Any]	) ->						List[str]:
   A_      ,						A_      ,						A_					:				Dict          =			pt_weights[f"{prefix}.in_proj_weight"].chunk(3				,				dim=0	)
   A_      ,						A_      ,						A_					:				Union[str, Any]          =			pt_weights[f"{prefix}.in_proj_bias"].chunk(3				,				dim=0	)

   A_					:				int          =			pt_weights[f"{prefix}.out_proj.weight"]
   A_					:				Dict          =			pt_weights[f"{prefix}.out_proj.bias"]

   A_					:				Optional[int]          =			q_proj
   A_					:				Optional[int]          =			q_proj_bias

   A_					:				Tuple          =			k_proj
   A_					:				str          =			k_proj_bias

   A_					:				Optional[int]          =			v_proj
   A_					:				int          =			v_proj_bias

   A_					:				int          =			out_proj_weights
   A_					:				List[Any]          =			out_proj_bias

def      __snake_case						(					_lowerCAmelCase : int				,				_lowerCAmelCase : Dict				,				_lowerCAmelCase : int	) ->						Dict:
   copy_linear(hf_mlp.fca				,				_lowerCAmelCase				,				f"{prefix}.c_fc"	)
   copy_linear(hf_mlp.fca				,				_lowerCAmelCase				,				f"{prefix}.c_proj"	)

def      __snake_case						(					_lowerCAmelCase : Tuple				,				_lowerCAmelCase : Optional[Any]				,				_lowerCAmelCase : Optional[int]	) ->						List[str]:
   A_					:				List[str]          =			pt_weights[f"{prefix}.weight"].data
   A_					:				int          =			pt_weights[f"{prefix}.bias"].data

def      __snake_case						(					_lowerCAmelCase : Dict				,				_lowerCAmelCase : str				,				_lowerCAmelCase : Optional[Any]	) ->						str:
   # copy layer norms
   copy_linear(hf_layer.layer_norma				,				_lowerCAmelCase				,				f"{prefix}.ln_1"	)
   copy_linear(hf_layer.layer_norma				,				_lowerCAmelCase				,				f"{prefix}.ln_2"	)

   # copy MLP
   copy_mlp(hf_layer.mlp				,				_lowerCAmelCase				,				f"{prefix}.mlp"	)

   # copy attn
   copy_attn_layer(hf_layer.self_attn				,				_lowerCAmelCase				,				f"{prefix}.attn"	)

def      __snake_case						(					_lowerCAmelCase : Optional[Any]				,				_lowerCAmelCase : int				,				_lowerCAmelCase : Optional[Any]	) ->						List[Any]:
   for layer_id, hf_layer in enumerate(_lowerCAmelCase	):
      copy_layer(_lowerCAmelCase				,				_lowerCAmelCase				,				f"{prefix}.{layer_id}"	)

def      __snake_case						(					_lowerCAmelCase : Union[str, Any]				,				_lowerCAmelCase : Any	) ->						Dict:
   # copy projection
   A_					:				Optional[Any]          =			pt_weights["text_projection"].data.T

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

def      __snake_case						(					_lowerCAmelCase : int				,				_lowerCAmelCase : Any	) ->						List[str]:
   # copy projection
   A_					:				Tuple          =			pt_weights["visual.proj"].data.T

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

   # copy embeddings
   A_					:				List[Any]          =			pt_weights["visual.conv1.weight"].data
   A_					:				Union[str, Any]          =			pt_weights["visual.class_embedding"].data
   A_					:				Tuple          =			pt_weights["visual.positional_embedding"].data

   # copy encoder
   copy_layers(hf_model.vision_model.encoder.layers				,				_lowerCAmelCase				,				"visual.transformer.resblocks"	)

@torch.no_grad()
def      __snake_case						(					_lowerCAmelCase : List[Any]				,				_lowerCAmelCase : int				,				_lowerCAmelCase : Optional[int]=None	) ->						str:
   assert config_path is not None, "Please specify the ChineseCLIP model config of the corresponding model size."
   A_					:				Optional[Any]          =			ChineseCLIPConfig.from_pretrained(_lowerCAmelCase	)

   A_					:				Tuple          =			ChineseCLIPModel(_lowerCAmelCase	).eval()

   A_					:				Any          =			torch.load(_lowerCAmelCase				,				map_location="cpu"	)["state_dict"]
   A_					:				Optional[Any]          =			{(name[7:] if name.startswith("module."	) else name): value for name, value in pt_weights.items()}

   copy_text_model_and_projection(_lowerCAmelCase				,				_lowerCAmelCase	)
   copy_vision_model_and_projection(_lowerCAmelCase				,				_lowerCAmelCase	)
   A_					:				List[Any]          =			pt_weights["logit_scale"].data

   hf_model.save_pretrained(_lowerCAmelCase	)


if __name__ == "__main__":
 _lowerCAmelCase							:      Optional[Any]						=	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.'''
 )
 _lowerCAmelCase							:      Tuple						=	parser.parse_args()

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