





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     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase					):
       _UpperCamelCase              =						'https://storage.googleapis.com/sfr-vision-language-research/BLIP/demo.jpg'
       _UpperCamelCase              =						Image.open(requests.get(lowerCAmelCase    ,       stream=lowerCAmelCase					).raw					).convert('''RGB'''					)

       _UpperCamelCase              =						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)					),
           ]					)
       _UpperCamelCase              =						transform(lowerCAmelCase					).unsqueeze(0					).to(lowerCAmelCase					)
       return image


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

       if "encoder.pos_embed" in key:
              _UpperCamelCase              =						re.sub(R'''encoder.pos_embed'''    ,       '''embeddings.position_embedding'''    ,       lowerCAmelCase					)
       if "encoder.cls_token" in key:
              _UpperCamelCase              =						re.sub(R'''encoder.cls_token'''    ,       '''embeddings.class_embedding'''    ,       lowerCAmelCase					)

       if "self_attn" in key:
              _UpperCamelCase              =						re.sub(R'''self_attn.proj'''    ,       '''self_attn.projection'''    ,       lowerCAmelCase					)

       return key


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

       _UpperCamelCase              =						BlipForConditionalGeneration(lowerCAmelCase					).eval()

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

       _UpperCamelCase              =						blip_decoder(pretrained=lowerCAmelCase    ,       image_size=384    ,       vit='''base'''					)
       _UpperCamelCase              =						pt_model.eval()

       _UpperCamelCase              =						pt_model.state_dict()
       for key in modified_state_dict.copy():
              _UpperCamelCase              =						modified_state_dict.pop(lowerCAmelCase					)
              _UpperCamelCase              =						rename_key(lowerCAmelCase					)
              _UpperCamelCase              =						value

       hf_model.load_state_dict(lowerCAmelCase					)

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

       _UpperCamelCase              =						hf_model.generate(lowerCAmelCase    ,       lowerCAmelCase					)

       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]

       _UpperCamelCase              =						hf_model.generate(lowerCAmelCase					)

       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(lowerCAmelCase					)

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

       _UpperCamelCase              =						blip_vqa(pretrained=lowerCAmelCase    ,       image_size=lowerCAmelCase    ,       vit='''base'''					)
       vqa_model.eval()

       _UpperCamelCase              =						vqa_model.state_dict()
       for key in modified_state_dict.copy():
              _UpperCamelCase              =						modified_state_dict.pop(lowerCAmelCase					)
              _UpperCamelCase              =						rename_key(lowerCAmelCase					)
              _UpperCamelCase              =						value

       _UpperCamelCase              =						BlipForQuestionAnswering(lowerCAmelCase					)

       hf_vqa_model.load_state_dict(lowerCAmelCase					)

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

       _UpperCamelCase              =						hf_vqa_model.generate(lowerCAmelCase    ,       lowerCAmelCase					)
       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'''					)

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

       _UpperCamelCase              =						blip_itm(pretrained=lowerCAmelCase    ,       image_size=lowerCAmelCase    ,       vit='''base'''					)
       itm_model.eval()

       _UpperCamelCase              =						itm_model.state_dict()
       for key in modified_state_dict.copy():
              _UpperCamelCase              =						modified_state_dict.pop(lowerCAmelCase					)
              _UpperCamelCase              =						rename_key(lowerCAmelCase					)
              _UpperCamelCase              =						value

       _UpperCamelCase              =						BlipForImageTextRetrieval(lowerCAmelCase					)

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

       hf_itm_model.load_state_dict(lowerCAmelCase					)
       hf_itm_model.eval()

       _UpperCamelCase              =						hf_itm_model(lowerCAmelCase    ,       lowerCAmelCase    ,       use_itm_head=lowerCAmelCase					)
       _UpperCamelCase              =						hf_itm_model(lowerCAmelCase    ,       lowerCAmelCase    ,       use_itm_head=lowerCAmelCase					)

       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__":
   lowercase    : str    	=  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""")
   lowercase    : Union[str, Any]    	=  parser.parse_args()

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













from __future__ import annotations

import math

import numpy as np
from numpy.linalg import norm


def     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase					):
       return math.sqrt(sum(pow(a - b    ,       2					) for a, b in zip(lowerCAmelCase    ,       lowerCAmelCase					)					)					)


def     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase					):

       if dataset.ndim != value_array.ndim:
              _UpperCamelCase              =						(
                  '''Wrong input data\'s dimensions... '''
                  f'''dataset : {dataset.ndim}, value_array : {value_array.ndim}'''
              )
              raise ValueError(lowerCAmelCase					)

       try:
              if dataset.shape[1] != value_array.shape[1]:
                     _UpperCamelCase              =						(
                         '''Wrong input data\'s shape... '''
                         f'''dataset : {dataset.shape[1]}, value_array : {value_array.shape[1]}'''
                     )
                     raise ValueError(lowerCAmelCase					)
       except IndexError:
              if dataset.ndim != value_array.ndim:
                     raise TypeError('''Wrong shape'''					)

       if dataset.dtype != value_array.dtype:
              _UpperCamelCase              =						(
                  '''Input data have different datatype... '''
                  f'''dataset : {dataset.dtype}, value_array : {value_array.dtype}'''
              )
              raise TypeError(lowerCAmelCase					)

       _UpperCamelCase              =						[]

       for value in value_array:
              _UpperCamelCase              =						euclidean(lowerCAmelCase    ,       dataset[0]					)
              _UpperCamelCase              =						dataset[0].tolist()

              for dataset_value in dataset[1:]:
                     _UpperCamelCase              =						euclidean(lowerCAmelCase    ,       lowerCAmelCase					)

                     if dist > temp_dist:
                            _UpperCamelCase              =						temp_dist
                            _UpperCamelCase              =						dataset_value.tolist()

              answer.append([vector, dist]					)

       return answer


def     SCREAMING_SNAKE_CASE			(      lowerCAmelCase    ,       lowerCAmelCase					):
       return np.dot(lowerCAmelCase    ,       lowerCAmelCase					) / (norm(lowerCAmelCase					) * norm(lowerCAmelCase					))


if __name__ == "__main__":
       import doctest

       doctest.testmod()






