
"""simple docstring"""




import os
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from huggingface_hub.file_download import http_get
from requests.exceptions import HTTPError

from transformers import (
    AlbertTokenizer,
    AutoTokenizer,
    BertTokenizer,
    BertTokenizerFast,
    GPTaTokenizerFast,
    is_tokenizers_available,
)
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_tokenizers
from transformers.tokenization_utils import Trie


sys.path.append(str(Path(__file__).parent.parent / 'utils'))

from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
						from test_module.custom_tokenization_fast import CustomTokenizerFast





class 			snake_case  (				unittest.TestCase ):

							def _lowercase       (self     ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_    			=			mock.Mock()
									SCREAMING_SNAKE_CASE_    			=			5_00
									SCREAMING_SNAKE_CASE_    			=			{}
									SCREAMING_SNAKE_CASE_    			=			HTTPError
									SCREAMING_SNAKE_CASE_    			=			{}

									# Download this model to make sure it's in the cache.
									SCREAMING_SNAKE_CASE_    			=			BertTokenizer.from_pretrained('''hf-internal-testing/tiny-random-bert'''     )

									# Under the mock environment we get a 500 error when trying to reach the tokenizer.
									with mock.patch('''requests.Session.request'''							,							return_value=SCREAMING_SNAKE_CASE_     ) as mock_head:
											SCREAMING_SNAKE_CASE_    			=			BertTokenizer.from_pretrained('''hf-internal-testing/tiny-random-bert'''     )
											# This check we did call the fake head request
											mock_head.assert_called()

							@require_tokenizers
							def _lowercase       (self     ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_    			=			mock.Mock()
									SCREAMING_SNAKE_CASE_    			=			5_00
									SCREAMING_SNAKE_CASE_    			=			{}
									SCREAMING_SNAKE_CASE_    			=			HTTPError
									SCREAMING_SNAKE_CASE_    			=			{}

									# Download this model to make sure it's in the cache.
									SCREAMING_SNAKE_CASE_    			=			GPTaTokenizerFast.from_pretrained('''gpt2'''     )

									# Under the mock environment we get a 500 error when trying to reach the tokenizer.
									with mock.patch('''requests.Session.request'''							,							return_value=SCREAMING_SNAKE_CASE_     ) as mock_head:
											SCREAMING_SNAKE_CASE_    			=			GPTaTokenizerFast.from_pretrained('''gpt2'''     )
											# This check we did call the fake head request
											mock_head.assert_called()

							def _lowercase       (self     ):

									"""simple docstring"""







									try:
											SCREAMING_SNAKE_CASE_    			=			tempfile.mktemp()
											with open(SCREAMING_SNAKE_CASE_							,							'''wb'''     ) as f:
													http_get('''https://huggingface.co/albert-base-v1/resolve/main/spiece.model'''							,							SCREAMING_SNAKE_CASE_     )

											SCREAMING_SNAKE_CASE_    			=			AlbertTokenizer.from_pretrained(SCREAMING_SNAKE_CASE_     )
									finally:
											os.remove(SCREAMING_SNAKE_CASE_     )

									# Supporting this legacy load introduced a weird bug where the tokenizer would load local files if they are in
									# the current folder and have the right name.
									if os.path.isfile('''tokenizer.json'''     ):
											# We skip the test if the user has a `tokenizer.json` in this folder to avoid deleting it.
											return
									try:
											with open('''tokenizer.json'''							,							'''wb'''     ) as f:
													http_get('''https://huggingface.co/hf-internal-testing/tiny-random-bert/blob/main/tokenizer.json'''							,							SCREAMING_SNAKE_CASE_     )
											SCREAMING_SNAKE_CASE_    			=			AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''     )
											# The tiny random BERT has a vocab size of 1024, tiny gpt2 as a vocab size of 1000
											self.assertEqual(tokenizer.vocab_size							,							10_00     )
											# Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

									finally:
											os.remove('''tokenizer.json'''     )







							def _lowercase       (self     ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_    			=			AlbertTokenizer.from_pretrained('''https://huggingface.co/albert-base-v1/resolve/main/spiece.model'''     )







@is_staging_test
class 			snake_case  (				unittest.TestCase ):
							UpperCAmelCase__      =   ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']

							@classmethod
							def _lowercase       (cls     ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_    			=			TOKEN
									HfFolder.save_token(SCREAMING_SNAKE_CASE_     )

							@classmethod
							def _lowercase       (cls     ):

									"""simple docstring"""







									try:
											delete_repo(token=cls._token							,							repo_id='''test-tokenizer'''     )
									except HTTPError:
											pass

									try:
											delete_repo(token=cls._token							,							repo_id='''valid_org/test-tokenizer-org'''     )
									except HTTPError:
											pass

									try:
											delete_repo(token=cls._token							,							repo_id='''test-dynamic-tokenizer'''     )
									except HTTPError:
											pass

							def _lowercase       (self     ):

									"""simple docstring"""







									with tempfile.TemporaryDirectory() as tmp_dir:
											SCREAMING_SNAKE_CASE_    			=			os.path.join(SCREAMING_SNAKE_CASE_							,							'''vocab.txt'''     )
											with open(SCREAMING_SNAKE_CASE_							,							'''w'''							,							encoding='''utf-8'''     ) as vocab_writer:
													vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]     )     )
											SCREAMING_SNAKE_CASE_    			=			BertTokenizer(SCREAMING_SNAKE_CASE_     )

									tokenizer.push_to_hub('''test-tokenizer'''							,							use_auth_token=self._token     )
									SCREAMING_SNAKE_CASE_    			=			BertTokenizer.from_pretrained(f'{USER}/test-tokenizer'     )
									self.assertDictEqual(new_tokenizer.vocab							,							tokenizer.vocab     )

									# Reset repo
									delete_repo(token=self._token							,							repo_id='''test-tokenizer'''     )

									# Push to hub via save_pretrained
									with tempfile.TemporaryDirectory() as tmp_dir:
											tokenizer.save_pretrained(SCREAMING_SNAKE_CASE_							,							repo_id='''test-tokenizer'''							,							push_to_hub=SCREAMING_SNAKE_CASE_							,							use_auth_token=self._token     )

									SCREAMING_SNAKE_CASE_    			=			BertTokenizer.from_pretrained(f'{USER}/test-tokenizer'     )
									self.assertDictEqual(new_tokenizer.vocab							,							tokenizer.vocab     )

							def _lowercase       (self     ):

									"""simple docstring"""







									with tempfile.TemporaryDirectory() as tmp_dir:
											SCREAMING_SNAKE_CASE_    			=			os.path.join(SCREAMING_SNAKE_CASE_							,							'''vocab.txt'''     )
											with open(SCREAMING_SNAKE_CASE_							,							'''w'''							,							encoding='''utf-8'''     ) as vocab_writer:
													vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]     )     )
											SCREAMING_SNAKE_CASE_    			=			BertTokenizer(SCREAMING_SNAKE_CASE_     )

									tokenizer.push_to_hub('''valid_org/test-tokenizer-org'''							,							use_auth_token=self._token     )
									SCREAMING_SNAKE_CASE_    			=			BertTokenizer.from_pretrained('''valid_org/test-tokenizer-org'''     )
									self.assertDictEqual(new_tokenizer.vocab							,							tokenizer.vocab     )

									# Reset repo
									delete_repo(token=self._token							,							repo_id='''valid_org/test-tokenizer-org'''     )

									# Push to hub via save_pretrained
									with tempfile.TemporaryDirectory() as tmp_dir:
											tokenizer.save_pretrained(
											    SCREAMING_SNAKE_CASE_							,							repo_id='''valid_org/test-tokenizer-org'''							,							push_to_hub=SCREAMING_SNAKE_CASE_							,							use_auth_token=self._token     )

									SCREAMING_SNAKE_CASE_    			=			BertTokenizer.from_pretrained('''valid_org/test-tokenizer-org'''     )
									self.assertDictEqual(new_tokenizer.vocab							,							tokenizer.vocab     )







							@require_tokenizers
							def _lowercase       (self     ):

									"""simple docstring"""







									CustomTokenizer.register_for_auto_class()
									with tempfile.TemporaryDirectory() as tmp_dir:
											SCREAMING_SNAKE_CASE_    			=			os.path.join(SCREAMING_SNAKE_CASE_							,							'''vocab.txt'''     )
											with open(SCREAMING_SNAKE_CASE_							,							'''w'''							,							encoding='''utf-8'''     ) as vocab_writer:
													vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]     )     )
											SCREAMING_SNAKE_CASE_    			=			CustomTokenizer(SCREAMING_SNAKE_CASE_     )

									# No fast custom tokenizer
									tokenizer.push_to_hub('''test-dynamic-tokenizer'''							,							use_auth_token=self._token     )

									SCREAMING_SNAKE_CASE_    			=			AutoTokenizer.from_pretrained(f'{USER}/test-dynamic-tokenizer'							,							trust_remote_code=SCREAMING_SNAKE_CASE_     )
									# Can't make an isinstance check because the new_model.config is from the CustomTokenizer class of a dynamic module
									self.assertEqual(tokenizer.__class__.__name__							,							'''CustomTokenizer'''     )

									# Fast and slow custom tokenizer
									CustomTokenizerFast.register_for_auto_class()
									with tempfile.TemporaryDirectory() as tmp_dir:
											SCREAMING_SNAKE_CASE_    			=			os.path.join(SCREAMING_SNAKE_CASE_							,							'''vocab.txt'''     )
											with open(SCREAMING_SNAKE_CASE_							,							'''w'''							,							encoding='''utf-8'''     ) as vocab_writer:
													vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]     )     )

											SCREAMING_SNAKE_CASE_    			=			BertTokenizerFast.from_pretrained(SCREAMING_SNAKE_CASE_     )
											bert_tokenizer.save_pretrained(SCREAMING_SNAKE_CASE_     )
											SCREAMING_SNAKE_CASE_    			=			CustomTokenizerFast.from_pretrained(SCREAMING_SNAKE_CASE_     )

									tokenizer.push_to_hub('''test-dynamic-tokenizer'''							,							use_auth_token=self._token     )

									SCREAMING_SNAKE_CASE_    			=			AutoTokenizer.from_pretrained(f'{USER}/test-dynamic-tokenizer'							,							trust_remote_code=SCREAMING_SNAKE_CASE_     )
									# Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
									self.assertEqual(tokenizer.__class__.__name__							,							'''CustomTokenizerFast'''     )
									SCREAMING_SNAKE_CASE_    			=			AutoTokenizer.from_pretrained(
									    f'{USER}/test-dynamic-tokenizer'							,							use_fast=SCREAMING_SNAKE_CASE_							,							trust_remote_code=SCREAMING_SNAKE_CASE_     )
									# Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
									self.assertEqual(tokenizer.__class__.__name__							,							'''CustomTokenizer'''     )







class 			snake_case  (				unittest.TestCase ):

							def _lowercase       (self     ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_    			=			Trie()
									trie.add('''Hello 友達'''     )
									self.assertEqual(trie.data							,							{'''H''': {'''e''': {'''l''': {'''l''': {'''o''': {''' ''': {'''友''': {'''達''': {'''''': 1}}}}}}}}}     )
									trie.add('''Hello'''     )
									trie.data
									self.assertEqual(trie.data							,							{'''H''': {'''e''': {'''l''': {'''l''': {'''o''': {'''''': 1, ''' ''': {'''友''': {'''達''': {'''''': 1}}}}}}}}}     )

							def _lowercase       (self     ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_    			=			Trie()
									self.assertEqual(trie.split('''[CLS] This is a extra_id_100'''     )							,							['''[CLS] This is a extra_id_100''']     )
									trie.add('''[CLS]'''     )
									trie.add('''extra_id_1'''     )
									trie.add('''extra_id_100'''     )
									self.assertEqual(trie.split('''[CLS] This is a extra_id_100'''     )							,							['''[CLS]''', ''' This is a ''', '''extra_id_100''']     )

							def _lowercase       (self     ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_    			=			Trie()
									trie.add('''A'''     )
									self.assertEqual(trie.split('''ABC'''     )							,							['''A''', '''BC''']     )
									self.assertEqual(trie.split('''BCA'''     )							,							['''BC''', '''A''']     )

							def _lowercase       (self     ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_    			=			Trie()
									trie.add('''TOKEN]'''     )
									trie.add('''[SPECIAL_TOKEN]'''     )
									self.assertEqual(trie.split('''This is something [SPECIAL_TOKEN]'''     )							,							['''This is something ''', '''[SPECIAL_TOKEN]''']     )

							def _lowercase       (self     ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_    			=			Trie()
									trie.add('''A'''     )
									trie.add('''P'''     )
									trie.add('''[SPECIAL_TOKEN]'''     )
									self.assertEqual(trie.split('''This is something [SPECIAL_TOKEN]'''     )							,							['''This is something ''', '''[SPECIAL_TOKEN]''']     )

							def _lowercase       (self     ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_    			=			Trie()
									trie.add('''AB'''     )
									trie.add('''B'''     )
									trie.add('''C'''     )
									self.assertEqual(trie.split('''ABC'''     )							,							['''AB''', '''C''']     )

							def _lowercase       (self     ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_    			=			Trie()
									trie.add('''ABC'''     )
									trie.add('''B'''     )
									trie.add('''CD'''     )
									self.assertEqual(trie.split('''ABCD'''     )							,							['''ABC''', '''D''']     )







							def _lowercase       (self     ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_    			=			Trie()
									SCREAMING_SNAKE_CASE_    			=			trie.cut_text('''ABC'''							,							[0, 0, 2, 1, 2, 3]     )
									self.assertEqual(SCREAMING_SNAKE_CASE_							,							['''AB''', '''C''']     )

"""simple docstring"""




import argparse
import re

import requests
import torch

# git clone https://github.com/salesforce/BLIP.git
from models.blip import blip_decoder
from models.blip_itm import blip_itm
from models.blip_vqa import blip_vqa
from PIL import Image
from torchvision import transforms
from torchvision.transforms.functional import InterpolationMode

from transformers import (
    BertTokenizer,
    BlipConfig,
    BlipForConditionalGeneration,
    BlipForImageTextRetrieval,
    BlipForQuestionAnswering,
)



def 							_lowerCamelCase			(			__a,				__a     ):
		SCREAMING_SNAKE_CASE_    			=			'''https://storage.googleapis.com/sfr-vision-language-research/BLIP/demo.jpg'''
		SCREAMING_SNAKE_CASE_    			=			Image.open(requests.get(__a,				stream=__a     ).raw     ).convert('''RGB'''     )

		SCREAMING_SNAKE_CASE_    			=			transforms.Compose(
		    [
		        transforms.Resize((image_size, image_size),				interpolation=InterpolationMode.BICUBIC     ),
		        transforms.ToTensor(),
		        transforms.Normalize((0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3),				(0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1)     ),
		    ]     )
		SCREAMING_SNAKE_CASE_    			=			transform(__a     ).unsqueeze(0     ).to(__a     )
		return image



def 							_lowerCamelCase			(			__a     ):
		if "visual_encoder" in key:
				SCREAMING_SNAKE_CASE_    			=			re.sub('''visual_encoder*''',				'''vision_model.encoder''',				__a     )
		if "blocks" in key:
				SCREAMING_SNAKE_CASE_    			=			re.sub(r'''blocks''',				'''layers''',				__a     )
		if "attn" in key:
				SCREAMING_SNAKE_CASE_    			=			re.sub(r'''attn''',				'''self_attn''',				__a     )
		if "norm1" in key:
				SCREAMING_SNAKE_CASE_    			=			re.sub(r'''norm1''',				'''layer_norm1''',				__a     )
		if "norm2" in key:
				SCREAMING_SNAKE_CASE_    			=			re.sub(r'''norm2''',				'''layer_norm2''',				__a     )
		if "encoder.norm" in key:
				SCREAMING_SNAKE_CASE_    			=			re.sub(r'''encoder.norm''',				'''post_layernorm''',				__a     )
		if "encoder.patch_embed.proj" in key:
				SCREAMING_SNAKE_CASE_    			=			re.sub(r'''encoder.patch_embed.proj''',				'''embeddings.patch_embedding''',				__a     )

		if "encoder.pos_embed" in key:
				SCREAMING_SNAKE_CASE_    			=			re.sub(r'''encoder.pos_embed''',				'''embeddings.position_embedding''',				__a     )
		if "encoder.cls_token" in key:
				SCREAMING_SNAKE_CASE_    			=			re.sub(r'''encoder.cls_token''',				'''embeddings.class_embedding''',				__a     )

		if "self_attn" in key:
				SCREAMING_SNAKE_CASE_    			=			re.sub(r'''self_attn.proj''',				'''self_attn.projection''',				__a     )

		return key



@torch.no_grad()
def 							_lowerCamelCase			(			__a,				__a=None     ):
		if config_path is not None:
				SCREAMING_SNAKE_CASE_    			=			BlipConfig.from_pretrained(__a     )
		else:
				SCREAMING_SNAKE_CASE_    			=			BlipConfig(projection_dim=512,				text_config={},				vision_config={}     )

		SCREAMING_SNAKE_CASE_    			=			BlipForConditionalGeneration(__a     ).eval()

		SCREAMING_SNAKE_CASE_    			=			'''https://storage.googleapis.com/sfr-vision-language-research/BLIP/models/model_base_capfilt_large.pth'''

		SCREAMING_SNAKE_CASE_    			=			blip_decoder(pretrained=__a,				image_size=384,				vit='''base'''     )
		SCREAMING_SNAKE_CASE_    			=			pt_model.eval()

		SCREAMING_SNAKE_CASE_    			=			pt_model.state_dict()
		for key in modified_state_dict.copy():
				SCREAMING_SNAKE_CASE_    			=			modified_state_dict.pop(__a     )
				SCREAMING_SNAKE_CASE_    			=			rename_key(__a     )
				SCREAMING_SNAKE_CASE_    			=			value

		hf_model.load_state_dict(__a     )

		SCREAMING_SNAKE_CASE_    			=			384
		SCREAMING_SNAKE_CASE_    			=			load_demo_image(image_size=__a,				device='''cpu'''     )
		SCREAMING_SNAKE_CASE_    			=			BertTokenizer.from_pretrained('''bert-base-uncased'''     )
		SCREAMING_SNAKE_CASE_    			=			tokenizer(['''a picture of''']     ).input_ids

		SCREAMING_SNAKE_CASE_    			=			hf_model.generate(__a,				__a     )

		assert out[0].tolist() == [30_522, 1_037, 3_861, 1_997, 1_037, 2_450, 3_564, 2_006, 1_996, 3_509, 2_007, 2_014, 3_899, 102]

		SCREAMING_SNAKE_CASE_    			=			hf_model.generate(__a     )

		assert out[0].tolist() == [30_522, 1_037, 2_450, 3_564, 2_006, 1_996, 3_509, 2_007, 2_014, 3_899, 102]

		if pytorch_dump_folder_path is not None:
				hf_model.save_pretrained(__a     )

		# model_url = 'https://storage.googleapis.com/sfr-vision-language-research/BLIP/models/model_vqa.pth'
		SCREAMING_SNAKE_CASE_    			=			(
		    '''https://storage.googleapis.com/sfr-vision-language-research/BLIP/models/model_base_vqa_capfilt_large.pth'''
		)

		SCREAMING_SNAKE_CASE_    			=			blip_vqa(pretrained=__a,				image_size=__a,				vit='''base'''     )
		vqa_model.eval()

		SCREAMING_SNAKE_CASE_    			=			vqa_model.state_dict()
		for key in modified_state_dict.copy():
				SCREAMING_SNAKE_CASE_    			=			modified_state_dict.pop(__a     )
				SCREAMING_SNAKE_CASE_    			=			rename_key(__a     )
				SCREAMING_SNAKE_CASE_    			=			value

		SCREAMING_SNAKE_CASE_    			=			BlipForQuestionAnswering(__a     )

		hf_vqa_model.load_state_dict(__a     )

		SCREAMING_SNAKE_CASE_    			=			['''How many dogs are in this image?''']
		SCREAMING_SNAKE_CASE_    			=			tokenizer(__a,				return_tensors='''pt'''     ).input_ids

		SCREAMING_SNAKE_CASE_    			=			hf_vqa_model.generate(__a,				__a     )
		print(tokenizer.decode(answer[0]     )     )

		assert tokenizer.decode(answer[0]     ) == "[UNK] 1 [SEP]"
		if pytorch_dump_folder_path is not None:
				hf_vqa_model.save_pretrained(pytorch_dump_folder_path + '''_vqa'''     )

		SCREAMING_SNAKE_CASE_    			=			'''https://storage.googleapis.com/sfr-vision-language-research/BLIP/models/model_base_retrieval_coco.pth'''

		SCREAMING_SNAKE_CASE_    			=			blip_itm(pretrained=__a,				image_size=__a,				vit='''base'''     )
		itm_model.eval()

		SCREAMING_SNAKE_CASE_    			=			itm_model.state_dict()
		for key in modified_state_dict.copy():
				SCREAMING_SNAKE_CASE_    			=			modified_state_dict.pop(__a     )
				SCREAMING_SNAKE_CASE_    			=			rename_key(__a     )
				SCREAMING_SNAKE_CASE_    			=			value

		SCREAMING_SNAKE_CASE_    			=			BlipForImageTextRetrieval(__a     )

		SCREAMING_SNAKE_CASE_    			=			['''A picture of a woman with a dog sitting in a beach''']
		SCREAMING_SNAKE_CASE_    			=			tokenizer(
		    __a,				return_tensors='''pt''',				padding='''max_length''',				truncation=__a,				max_length=35,				).input_ids

		hf_itm_model.load_state_dict(__a     )
		hf_itm_model.eval()

		SCREAMING_SNAKE_CASE_    			=			hf_itm_model(__a,				__a,				use_itm_head=__a     )
		SCREAMING_SNAKE_CASE_    			=			hf_itm_model(__a,				__a,				use_itm_head=__a     )

		assert out[0].item() == 0.2_1_1_0_6_8_7_4_9_4_2_7_7_9_5_4
		assert torch.nn.functional.softmax(out_itm[0],				dim=1     )[:, 1].item() == 0.4_5_6_9_8_8_4_5_3_8_6_5_0_5_1_2_7

		if pytorch_dump_folder_path is not None:
				hf_itm_model.save_pretrained(pytorch_dump_folder_path + '''_itm'''     )


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('--config_path', default=None, type=str, help='Path to hf config.json of model to convert')
						lowerCAmelCase__				=					parser.parse_args()

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