


import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


if TYPE_CHECKING:
  from transformers.pipelines.conversational import Conversation

UpperCAmelCase__        =				logging.get_logger(__name__)

UpperCAmelCase__        =				{
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
}

UpperCAmelCase__        =				{
    '''vocab_file''': {
        '''gpt2''': '''https://huggingface.co/gpt2/resolve/main/vocab.json''',
        '''gpt2-medium''': '''https://huggingface.co/gpt2-medium/resolve/main/vocab.json''',
        '''gpt2-large''': '''https://huggingface.co/gpt2-large/resolve/main/vocab.json''',
        '''gpt2-xl''': '''https://huggingface.co/gpt2-xl/resolve/main/vocab.json''',
        '''distilgpt2''': '''https://huggingface.co/distilgpt2/resolve/main/vocab.json''',
    },
    '''merges_file''': {
        '''gpt2''': '''https://huggingface.co/gpt2/resolve/main/merges.txt''',
        '''gpt2-medium''': '''https://huggingface.co/gpt2-medium/resolve/main/merges.txt''',
        '''gpt2-large''': '''https://huggingface.co/gpt2-large/resolve/main/merges.txt''',
        '''gpt2-xl''': '''https://huggingface.co/gpt2-xl/resolve/main/merges.txt''',
        '''distilgpt2''': '''https://huggingface.co/distilgpt2/resolve/main/merges.txt''',
    },
}

UpperCAmelCase__        =				{
    '''gpt2''': 1024,
    '''gpt2-medium''': 1024,
    '''gpt2-large''': 1024,
    '''gpt2-xl''': 1024,
    '''distilgpt2''': 1024,
}
@lru_cache()
def 							UpperCAmelCase_   (				)	->     Optional[int]:







 """simple docstring"""


 _lowercase				 =(
     list(range(ord('''!'''		)	,		ord('''~'''		) + 1		)		) + list(range(ord('''¡'''		)	,		ord('''¬'''		) + 1		)		) + list(range(ord('''®'''		)	,		ord('''ÿ'''		) + 1		)		)
 )
 _lowercase				 =bs[:]
 _lowercase				 =0
 for b in range(2**8		):
  if b not in bs:
   bs.append(UpperCAmelCase_		)
   cs.append(2**8 + n		)
   n += 1
 _lowercase				 =[chr(UpperCAmelCase_		) for n in cs]
 return dict(zip(UpperCAmelCase_	,		UpperCAmelCase_		)		)
def 							UpperCAmelCase_   (				__snake_case		)	->     Dict:







 """simple docstring"""


 _lowercase				 =set()
 _lowercase				 =word[0]
 for char in word[1:]:
  pairs.add((prev_char, char)		)
  _lowercase				 =char
 return pairs







class  lowerCamelCase__  (			_lowercase):
     SCREAMING_SNAKE_CASE__      =     VOCAB_FILES_NAMES
     SCREAMING_SNAKE_CASE__      =     PRETRAINED_VOCAB_FILES_MAP
     SCREAMING_SNAKE_CASE__      =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     SCREAMING_SNAKE_CASE__      =     ['''input_ids''', '''attention_mask''']





     def __init__(self						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase="replace"						,	UpperCAmelCase="<|endoftext|>"						,	UpperCAmelCase="<|endoftext|>"						,	UpperCAmelCase="<|endoftext|>"						,	UpperCAmelCase=None						,	UpperCAmelCase=False						,	UpperCAmelCase=False						,	**UpperCAmelCase						,	)	->  int:
      _lowercase				 =AddedToken(__snake_case						,	lstrip=__snake_case						,	rstrip=__snake_case			) if isinstance(__snake_case						,	__snake_case			) else bos_token
      _lowercase				 =AddedToken(__snake_case						,	lstrip=__snake_case						,	rstrip=__snake_case			) if isinstance(__snake_case						,	__snake_case			) else eos_token
      _lowercase				 =AddedToken(__snake_case						,	lstrip=__snake_case						,	rstrip=__snake_case			) if isinstance(__snake_case						,	__snake_case			) else unk_token
      _lowercase				 =AddedToken(__snake_case						,	lstrip=__snake_case						,	rstrip=__snake_case			) if isinstance(__snake_case						,	__snake_case			) else pad_token
      super().__init__(
          errors=__snake_case						,	unk_token=__snake_case						,	bos_token=__snake_case						,	eos_token=__snake_case						,	pad_token=__snake_case						,	add_prefix_space=__snake_case						,	add_bos_token=__snake_case						,	**__snake_case						,	)
      _lowercase				 =add_bos_token

      with open(__snake_case						,	encoding='''utf-8'''			) as vocab_handle:
       _lowercase				 =json.load(__snake_case			)
      _lowercase				 ={v: k for k, v in self.encoder.items()}
      _lowercase				 =errors  # how to handle errors in decoding
      _lowercase				 =bytes_to_unicode()
      _lowercase				 ={v: k for k, v in self.byte_encoder.items()}
      with open(__snake_case						,	encoding='''utf-8'''			) as merges_handle:
       _lowercase				 =merges_handle.read().split('''\n'''			)[1:-1]
      _lowercase				 =[tuple(merge.split()			) for merge in bpe_merges]
      _lowercase				 =dict(zip(__snake_case						,	range(len(__snake_case			)			)			)			)
      _lowercase				 ={}
      _lowercase				 =add_prefix_space

      # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
      _lowercase				 =re.compile(R'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''			)





     @property
     def 	__A   (self			)	->  Optional[Any]:
      return len(self.encoder			)





     def 	__A   (self			)	->  Dict:
      return dict(self.encoder						,	**self.added_tokens_encoder			)





     def 	__A   (self						,	UpperCAmelCase			)	->  List[str]:
      if token in self.cache:
       return self.cache[token]
      _lowercase				 =tuple(__snake_case			)
      _lowercase				 =get_pairs(__snake_case			)

      if not pairs:
       return token

      while True:
       _lowercase				 =min(__snake_case						,	key=lambda UpperCAmelCase			: self.bpe_ranks.get(__snake_case						,	float('''inf'''			)			)			)
       if bigram not in self.bpe_ranks:
        break
       _lowercase				 =bigram
       _lowercase				 =[]
       _lowercase				 =0
       while i < len(__snake_case			):
        try:
         _lowercase				 =word.index(__snake_case						,	__snake_case			)
        except ValueError:
         new_word.extend(word[i:]			)
         break
        else:
         new_word.extend(word[i:j]			)
         _lowercase				 =j

        if word[i] == first and i < len(__snake_case			) - 1 and word[i + 1] == second:
         new_word.append(first + second			)
         i += 2
        else:
         new_word.append(word[i]			)
         i += 1
       _lowercase				 =tuple(__snake_case			)
       _lowercase				 =new_word
       if len(__snake_case			) == 1:
        break
       else:
        _lowercase				 =get_pairs(__snake_case			)
      _lowercase				 =' '.join(__snake_case			)
      _lowercase				 =word
      return word





     def 	__A   (self						,	UpperCAmelCase						,	UpperCAmelCase=None			)	->  Any:
      if self.add_bos_token:
       _lowercase				 =[self.bos_token_id]
      else:
       _lowercase				 =[]

      _lowercase				 =bos_token_ids + token_ids_a

      if token_ids_a is None:
       return output

      return output + bos_token_ids + token_ids_a





     def 	__A   (self						,	UpperCAmelCase						,	UpperCAmelCase = None						,	UpperCAmelCase = False			)	->  List[str]:
      if already_has_special_tokens:
       return super().get_special_tokens_mask(
           token_ids_a=__snake_case						,	token_ids_a=__snake_case						,	already_has_special_tokens=__snake_case			)

      if not self.add_bos_token:
       return super().get_special_tokens_mask(
           token_ids_a=__snake_case						,	token_ids_a=__snake_case						,	already_has_special_tokens=__snake_case			)

      if token_ids_a is None:
       return [1] + ([0] * len(__snake_case			))
      return [1] + ([0] * len(__snake_case			)) + [1] + ([0] * len(__snake_case			))





     def 	__A   (self						,	UpperCAmelCase			)	->  str:
      _lowercase				 =[]
      for token in re.findall(self.pat						,	__snake_case			):
       _lowercase				 =''.join(
           self.byte_encoder[b] for b in token.encode('''utf-8'''			)			)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
       bpe_tokens.extend(bpe_token for bpe_token in self.bpe(__snake_case			).split(''' '''			)			)
      return bpe_tokens





     def 	__A   (self						,	UpperCAmelCase			)	->  Tuple:
      return self.encoder.get(__snake_case						,	self.encoder.get(self.unk_token			)			)





     def 	__A   (self						,	UpperCAmelCase			)	->  Dict:
      return self.decoder.get(__snake_case			)





     def 	__A   (self						,	UpperCAmelCase			)	->  Union[str, Any]:
      _lowercase				 =''.join(__snake_case			)
      _lowercase				 =bytearray([self.byte_decoder[c] for c in text]			).decode('''utf-8'''						,	errors=self.errors			)
      return text





     def 	__A   (self						,	UpperCAmelCase						,	UpperCAmelCase = None			)	->  Optional[Any]:
      if not os.path.isdir(__snake_case			):
       logger.error(f"Vocabulary path ({save_directory}) should be a directory"			)
       return
      _lowercase				 =os.path.join(
          __snake_case						,	(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']			)
      _lowercase				 =os.path.join(
          __snake_case						,	(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']			)

      with open(__snake_case						,	'''w'''						,	encoding='''utf-8'''			) as f:
       f.write(json.dumps(self.encoder						,	indent=2						,	sort_keys=__snake_case						,	ensure_ascii=__snake_case			) + '''\n'''			)

      _lowercase				 =0
      with open(__snake_case						,	'''w'''						,	encoding='''utf-8'''			) as writer:
       writer.write('''#version: 0.2\n'''			)
       for bpe_tokens, token_index in sorted(self.bpe_ranks.items()						,	key=lambda UpperCAmelCase			: kv[1]			):
        if index != token_index:
         logger.warning(
             f"Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."
             ''' Please check that the tokenizer is not corrupted!'''			)
         _lowercase				 =token_index
        writer.write(''' '''.join(__snake_case			) + '''\n'''			)
        index += 1

      return vocab_file, merge_file





     def 	__A   (self						,	UpperCAmelCase						,	UpperCAmelCase=False						,	**UpperCAmelCase			)	->  Optional[Any]:
      _lowercase				 =kwargs.pop('''add_prefix_space'''						,	self.add_prefix_space			)
      if is_split_into_words or add_prefix_space:
       _lowercase				 =' ' + text
      return (text, kwargs)





     def 	__A   (self						,	UpperCAmelCase			)	->  List[Any]:
      _lowercase				 =[]
      for is_user, text in conversation.iter_texts():
       input_ids.extend(self.encode(__snake_case						,	add_special_tokens=__snake_case			) + [self.eos_token_id]			)
      if len(__snake_case			) > self.model_max_length:
       _lowercase				 =input_ids[-self.model_max_length :]
      return input_ids




import argparse

import torch
from clip import load

from transformers import CLIPConfig, CLIPModel






def        lowerCamelCase_  (				UpperCAmelCase_				:  Optional[int]				,							UpperCAmelCase_				:  Tuple						)    ->					str:




	'''simple docstring'''




	_UpperCamelCase						,		_UpperCamelCase						,		_UpperCamelCase :		Tuple								=    pt_attn_layer.in_proj_weight.chunk(3				,							dim=0						)
	_UpperCamelCase						,		_UpperCamelCase						,		_UpperCamelCase :		str								=    pt_attn_layer.in_proj_bias.chunk(3				,							dim=0						)

	_UpperCamelCase :		Dict								=    pt_attn_layer.out_proj.weight
	_UpperCamelCase :		Tuple								=    pt_attn_layer.out_proj.bias

	_UpperCamelCase :		Tuple								=    q_proj
	_UpperCamelCase :		Dict								=    q_proj_bias

	_UpperCamelCase :		Dict								=    k_proj
	_UpperCamelCase :		str								=    k_proj_bias

	_UpperCamelCase :		Dict								=    v_proj
	_UpperCamelCase :		Optional[int]								=    v_proj_bias

	_UpperCamelCase :		Any								=    out_proj_weights
	_UpperCamelCase :		Optional[int]								=    out_proj_bias






def        lowerCamelCase_  (				UpperCAmelCase_				:  Optional[int]				,							UpperCAmelCase_				:  Union[str, Any]						)    ->					List[Any]:




	'''simple docstring'''




	copy_linear(hf_mlp.fca				,							pt_mlp.c_fc						)
	copy_linear(hf_mlp.fca				,							pt_mlp.c_proj						)






def        lowerCamelCase_  (				UpperCAmelCase_				:  List[str]				,							UpperCAmelCase_				:  List[str]						)    ->					List[Any]:




	'''simple docstring'''




	_UpperCamelCase :		int								=    pt_linear.weight
	_UpperCamelCase :		int								=    pt_linear.bias






def        lowerCamelCase_  (				UpperCAmelCase_				:  str				,							UpperCAmelCase_				:  Union[str, Any]						)    ->					Dict:




	'''simple docstring'''




	copy_linear(hf_layer.layer_norma				,							pt_layer.ln_a						)
	copy_linear(hf_layer.layer_norma				,							pt_layer.ln_a						)

	# copy MLP
	copy_mlp(hf_layer.mlp				,							pt_layer.mlp						)

	# copy attn
	copy_attn_layer(hf_layer.self_attn				,							pt_layer.attn						)






def        lowerCamelCase_  (				UpperCAmelCase_				:  Optional[Any]				,							UpperCAmelCase_				:  List[str]						)    ->					Dict:




	'''simple docstring'''




	for hf_layer, pt_layer in zip(UpperCAmelCase_				,							UpperCAmelCase_						):
		copy_layer(UpperCAmelCase_				,							UpperCAmelCase_						)






def        lowerCamelCase_  (				UpperCAmelCase_				:  Any				,							UpperCAmelCase_				:  Optional[Any]						)    ->					int:




	'''simple docstring'''




	_UpperCamelCase :		Optional[int]								=    pt_model.token_embedding.weight
	_UpperCamelCase :		Tuple								=    pt_model.positional_embedding

	# copy layer norm
	copy_linear(hf_encoder.final_layer_norm				,							pt_model.ln_final						)

	# copy hidden layers
	copy_layers(hf_encoder.encoder.layers				,							pt_model.transformer.resblocks						)






def        lowerCamelCase_  (				UpperCAmelCase_				:  int				,							UpperCAmelCase_				:  Tuple						)    ->					Union[str, Any]:




	'''simple docstring'''




	_UpperCamelCase :		Any								=    pt_model.text_projection.data.T

	# copy text encoder
	copy_encoder(hf_model.text_model				,							UpperCAmelCase_						)






def        lowerCamelCase_  (				UpperCAmelCase_				:  str				,							UpperCAmelCase_				:  Dict						)    ->					Optional[int]:




	'''simple docstring'''




	_UpperCamelCase :		str								=    pt_model.visual.proj.data.T

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

	# copy embeds
	_UpperCamelCase :		Tuple								=    pt_model.visual.conva.weight.data
	_UpperCamelCase :		Dict								=    pt_model.visual.class_embedding
	_UpperCamelCase :		str								=    pt_model.visual.positional_embedding.data

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






@torch.no_grad()
def        lowerCamelCase_  (				UpperCAmelCase_				:  Dict				,							UpperCAmelCase_				:  str				,							UpperCAmelCase_				:  List[str]=None						)    ->					Union[str, Any]:




	'''simple docstring'''




	if config_path is not None:
		_UpperCamelCase :		Dict								=    CLIPConfig.from_pretrained(UpperCAmelCase_						)
	else:
		_UpperCamelCase :		Tuple								=    CLIPConfig(projection_dim=5_1_2				,							text_config={}				,							vision_config={}						)

	_UpperCamelCase :		Tuple								=    CLIPModel(UpperCAmelCase_						).eval()

	_UpperCamelCase						,		_UpperCamelCase :		Optional[Any]								=    load(UpperCAmelCase_				,							device='cpu'				,							jit=UpperCAmelCase_						)
	_UpperCamelCase :		Optional[Any]								=    pt_model.eval()

	copy_text_model_and_projection(UpperCAmelCase_				,							UpperCAmelCase_						)
	copy_vison_model_and_projection(UpperCAmelCase_				,							UpperCAmelCase_						)
	_UpperCamelCase :		Optional[Any]								=    pt_model.logit_scale

	_UpperCamelCase :		Optional[int]								=    torch.arange(0				,							7_7						).unsqueeze(0						)
	_UpperCamelCase :		Dict								=    torch.randn(1				,							3				,							2_2_4				,							2_2_4						)

	_UpperCamelCase :		Optional[Any]								=    hf_model(input_ids=UpperCAmelCase_				,							pixel_values=UpperCAmelCase_				,							return_dict=UpperCAmelCase_						)
	_UpperCamelCase :		Optional[Any]								=    hf_outputs.logits_per_image
	_UpperCamelCase :		Optional[int]								=    hf_outputs.logits_per_text
	_UpperCamelCase						,		_UpperCamelCase :		int								=    pt_model(UpperCAmelCase_				,							UpperCAmelCase_						)

	assert torch.allclose(UpperCAmelCase_				,							UpperCAmelCase_				,							atol=1e-3						)
	assert torch.allclose(UpperCAmelCase_				,							UpperCAmelCase_				,							atol=1e-3						)

	hf_model.save_pretrained(UpperCAmelCase_						)


if __name__ == "__main__":
							lowerCAmelCase__             =		argparse.ArgumentParser()
							parser.add_argument("""--pytorch_dump_folder_path""", default=None, type=str, help="""Path to the output PyTorch model.""")
							parser.add_argument("""--checkpoint_path""", default=None, type=str, help="""Path to fairseq checkpoint""")
							parser.add_argument("""--config_path""", default=None, type=str, help="""Path to hf config.json of model to convert""")
							lowerCAmelCase__             =		parser.parse_args()

							convert_clip_checkpoint(args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path)




