




import argparse

import torch

from transformers import ChineseCLIPConfig, ChineseCLIPModel


def   A		(__A  :       List[Any]  ,							__A  :       List[str]  ,							__A  :       str			)  -> List[str]:







     """simple docstring"""


     UpperCAmelCase_		=				pt_weights[F"""{prefix}.in_proj_weight"""].chunk(3  ,							dim=0			)
     UpperCAmelCase_		=				pt_weights[F"""{prefix}.in_proj_bias"""].chunk(3  ,							dim=0			)

     UpperCAmelCase_		=				pt_weights[F"""{prefix}.out_proj.weight"""]
     UpperCAmelCase_		=				pt_weights[F"""{prefix}.out_proj.bias"""]

     UpperCAmelCase_		=				q_proj
     UpperCAmelCase_		=				q_proj_bias

     UpperCAmelCase_		=				k_proj
     UpperCAmelCase_		=				k_proj_bias

     UpperCAmelCase_		=				v_proj
     UpperCAmelCase_		=				v_proj_bias

     UpperCAmelCase_		=				out_proj_weights
     UpperCAmelCase_		=				out_proj_bias


def   A		(__A  :       Union[str, Any]  ,							__A  :       Optional[Any]  ,							__A  :       Any			)  -> List[str]:







     """simple docstring"""


     copy_linear(hf_mlp.fca  ,							a_  ,							F"""{prefix}.c_fc"""			)
     copy_linear(hf_mlp.fca  ,							a_  ,							F"""{prefix}.c_proj"""			)


def   A		(__A  :       Tuple  ,							__A  :       Optional[Any]  ,							__A  :       Optional[Any]			)  -> Optional[Any]:







     """simple docstring"""


     UpperCAmelCase_		=				pt_weights[F"""{prefix}.weight"""].data
     UpperCAmelCase_		=				pt_weights[F"""{prefix}.bias"""].data


def   A		(__A  :       Any  ,							__A  :       List[Any]  ,							__A  :       List[str]			)  -> List[str]:







     """simple docstring"""


     copy_linear(hf_layer.layer_norma  ,							a_  ,							F"""{prefix}.ln_1"""			)
     copy_linear(hf_layer.layer_norma  ,							a_  ,							F"""{prefix}.ln_2"""			)

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

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


def   A		(__A  :       str  ,							__A  :       List[str]  ,							__A  :       str			)  -> str:







     """simple docstring"""


     for layer_id, hf_layer in enumerate(a_			):
          copy_layer(a_  ,							a_  ,							F"""{prefix}.{layer_id}"""			)


def   A		(__A  :       int  ,							__A  :       Tuple			)  -> List[str]:







     """simple docstring"""


     UpperCAmelCase_		=				pt_weights['text_projection'].data.T

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


def   A		(__A  :       List[str]  ,							__A  :       int			)  -> Any:







     """simple docstring"""


     UpperCAmelCase_		=				pt_weights['visual.proj'].data.T

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

     # copy embeddings
     UpperCAmelCase_		=				pt_weights['visual.conv1.weight'].data
     UpperCAmelCase_		=				pt_weights['visual.class_embedding'].data
     UpperCAmelCase_		=				pt_weights['visual.positional_embedding'].data

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


@torch.no_grad()
def   A		(__A  :       Optional[Any]  ,							__A  :       int  ,							__A  :       Union[str, Any]=None			)  -> Tuple:







     """simple docstring"""


     assert config_path is not None, "Please specify the ChineseCLIP model config of the corresponding model size."
     UpperCAmelCase_		=				ChineseCLIPConfig.from_pretrained(a_			)

     UpperCAmelCase_		=				ChineseCLIPModel(a_			).eval()

     UpperCAmelCase_		=				torch.load(a_  ,							map_location='''cpu'''			)['state_dict']
     UpperCAmelCase_		=				{(name[7:] if name.startswith('''module.'''			) else name): value for name, value in pt_weights.items()}

     copy_text_model_and_projection(a_  ,							a_			)
     copy_vision_model_and_projection(a_  ,							a_			)
     UpperCAmelCase_		=				pt_weights['logit_scale'].data

     hf_model.save_pretrained(a_			)


if __name__ == "__main__":
    snake_case_					:					str       						=			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."
    )
    snake_case_					:					Any       						=			parser.parse_args()

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








"""simple docstring"""

import os
from typing import Optional, Union

import numpy as np
from huggingface_hub import hf_hub_download

from ... import AutoTokenizer
from ...utils import logging


_A				=					'block_records.npy'


_A				=					logging.get_logger(__name__)
def 						UpperCAmelCase   (						a_,		a_     ):





		'''simple docstring'''


		import tensorflow.compat.va as tf

		lowerCamelCase	:     Union[str, Any]            =       tf.data.TFRecordDataset(a_,		buffer_size=512 * 1024 * 1024     )
		lowerCamelCase	:     Union[str, Any]            =       blocks_dataset.batch(a_,		drop_remainder=a_     )
		lowerCamelCase	:     Union[str, Any]            =       next(blocks_dataset.take(1     ).as_numpy_iterator()     )

		return np_record






class 			_lowercase  :




	def __init__(					self						,					UpperCAmelCase_						,					UpperCAmelCase_						,					UpperCAmelCase_=2						,					UpperCAmelCase_=1000						,					UpperCAmelCase_=100						,					UpperCAmelCase_=100000						,					)						->				List[Any]:
			from scann.scann_ops.py.scann_ops_pybind import builder as Builder

			lowerCamelCase	:     int            =       Builder(db=UpperCAmelCase_						,					num_neighbors=UpperCAmelCase_						,					distance_measure='dot_product'   )
			lowerCamelCase	:     Optional[int]            =       builder.tree(
			    num_leaves=UpperCAmelCase_						,					num_leaves_to_search=UpperCAmelCase_						,					training_sample_size=UpperCAmelCase_   )
			lowerCamelCase	:     Union[str, Any]            =       builder.score_ah(dimensions_per_block=UpperCAmelCase_   )

			lowerCamelCase	:     List[Any]            =       builder.build()




	def      _UpperCamelCase       (					self						,					UpperCAmelCase_   )						->				Optional[int]:
			lowerCamelCase      ,       lowerCamelCase	:     List[Any]            =       self.searcher.search_batched(question_projection.detach().cpu()   )
			return retrieved_block_ids.astype('int64'   )






class 			_lowercase  :




	def __init__(					self						,					UpperCAmelCase_						,					UpperCAmelCase_   )						->				Optional[Any]:
			super().__init__()
			lowerCamelCase	:     Tuple            =       block_records
			lowerCamelCase	:     Any            =       tokenizer




	def __call__(					self						,					UpperCAmelCase_						,					UpperCAmelCase_						,					UpperCAmelCase_						,					UpperCAmelCase_=None						,					UpperCAmelCase_="pt"   )						->				Tuple:
			lowerCamelCase	:     List[str]            =       np.take(self.block_records						,					indices=UpperCAmelCase_						,					axis=0   )

			lowerCamelCase	:     Any            =       self.tokenizer.decode(question_input_ids[0]						,					skip_special_tokens=UpperCAmelCase_   )

			lowerCamelCase	:     int            =       []
			lowerCamelCase	:     Optional[int]            =       []
			for retrieved_block in retrieved_blocks:
					text.append(UpperCAmelCase_   )
					text_pair.append(retrieved_block.decode()   )

			lowerCamelCase	:     Union[str, Any]            =       self.tokenizer(
			    UpperCAmelCase_						,					UpperCAmelCase_						,					padding=UpperCAmelCase_						,					truncation=UpperCAmelCase_						,					return_special_tokens_mask=UpperCAmelCase_						,					max_length=UpperCAmelCase_   )
			lowerCamelCase	:     str            =       concat_inputs.convert_to_tensors(UpperCAmelCase_   )

			if answer_ids is not None:
					return self.block_has_answer(UpperCAmelCase_						,					UpperCAmelCase_   ) + (concat_inputs_tensors,)
			else:
					return (None, None, None, concat_inputs_tensors)




	@classmethod
	def      _UpperCamelCase       (					cls						,					UpperCAmelCase_						,					*UpperCAmelCase_						,					**UpperCAmelCase_   )						->				Dict:
			if os.path.isdir(UpperCAmelCase_   ):
					lowerCamelCase	:     str            =       os.path.join(UpperCAmelCase_						,					_REALM_BLOCK_RECORDS_FILENAME   )
			else:
					lowerCamelCase	:     Tuple            =       hf_hub_download(
					    repo_id=UpperCAmelCase_						,					filename=_REALM_BLOCK_RECORDS_FILENAME						,					**UpperCAmelCase_   )
			lowerCamelCase	:     Dict            =       np.load(UpperCAmelCase_						,					allow_pickle=UpperCAmelCase_   )

			lowerCamelCase	:     Union[str, Any]            =       AutoTokenizer.from_pretrained(UpperCAmelCase_						,					*UpperCAmelCase_						,					**UpperCAmelCase_   )

			return cls(UpperCAmelCase_						,					UpperCAmelCase_   )




	def      _UpperCamelCase       (					self						,					UpperCAmelCase_   )						->				Union[str, Any]:
			# save block records
			np.save(os.path.join(UpperCAmelCase_						,					_REALM_BLOCK_RECORDS_FILENAME   )						,					self.block_records   )
			# save tokenizer
			self.tokenizer.save_pretrained(UpperCAmelCase_   )




	def      _UpperCamelCase       (					self						,					UpperCAmelCase_						,					UpperCAmelCase_   )						->				Tuple:
			lowerCamelCase	:     List[Any]            =       []
			lowerCamelCase	:     Any            =       []
			lowerCamelCase	:     int            =       []
			lowerCamelCase	:     Union[str, Any]            =       0

			for input_id in concat_inputs.input_ids:
					lowerCamelCase	:     Optional[Any]            =       input_id.tolist()
					# Check answers between two [SEP] tokens
					lowerCamelCase	:     Tuple            =       input_id_list.index(self.tokenizer.sep_token_id   )
					lowerCamelCase	:     Any            =       first_sep_idx + 1 + input_id_list[first_sep_idx + 1 :].index(self.tokenizer.sep_token_id   )

					start_pos.append([]   )
					end_pos.append([]   )
					for answer in answer_ids:
							for idx in range(first_sep_idx + 1						,					UpperCAmelCase_   ):
									if answer[0] == input_id_list[idx]:
											if input_id_list[idx : idx + len(UpperCAmelCase_   )] == answer:
													start_pos[-1].append(UpperCAmelCase_   )
													end_pos[-1].append(idx + len(UpperCAmelCase_   ) - 1   )

					if len(start_pos[-1]   ) == 0:
							has_answers.append(UpperCAmelCase_   )
					else:
							has_answers.append(UpperCAmelCase_   )
							if len(start_pos[-1]   ) > max_answers:
									lowerCamelCase	:     List[str]            =       len(start_pos[-1]   )

        # Pad -1 to max_answers
			for start_pos_, end_pos_ in zip(UpperCAmelCase_						,					UpperCAmelCase_   ):
					if len(start_pos_   ) < max_answers:
							lowerCamelCase	:     Optional[int]            =       [-1] * (max_answers - len(start_pos_   ))
							start_pos_ += padded
							end_pos_ += padded
			return has_answers, start_pos, end_pos






