





import argparse

import requests
import torch

# pip3 install salesforce-lavis
# I'm actually installing a slightly modified version: pip3 install git+https://github.com/nielsrogge/LAVIS.git@fix_lavis
from lavis.models import load_model_and_preprocess
from PIL import Image

from transformers import (
    AutoTokenizer,
    BlipaConfig,
    BlipaForConditionalGeneration,
    BlipaProcessor,
    BlipaVisionConfig,
    BlipImageProcessor,
    OPTConfig,
    TaConfig,
)
from transformers.utils.constants import OPENAI_CLIP_MEAN, OPENAI_CLIP_STD


def 		_SCREAMING_SNAKE_CASE     ( ):
		lowerCamelCase_	:     Optional[int] 						=      'https://storage.googleapis.com/sfr-vision-language-research/LAVIS/assets/merlion.png'
		lowerCamelCase_	:     Union[str, Any] 						=      Image.open(requests.get(lowerCAmelCase__							,stream=lowerCAmelCase__						).raw						).convert('RGB'						)

		return image


def 		_SCREAMING_SNAKE_CASE     ( lowerCAmelCase__						):
		lowerCamelCase_	:     Tuple 						=      []
		# fmt: off

		# vision encoder
		rename_keys.append(('visual_encoder.cls_token', 'vision_model.embeddings.class_embedding')						)
		rename_keys.append(('visual_encoder.pos_embed', 'vision_model.embeddings.position_embedding')						)
		rename_keys.append(('visual_encoder.patch_embed.proj.weight', 'vision_model.embeddings.patch_embedding.weight')						)
		rename_keys.append(('visual_encoder.patch_embed.proj.bias', 'vision_model.embeddings.patch_embedding.bias')						)
		rename_keys.append(('ln_vision.weight', 'vision_model.post_layernorm.weight')						)
		rename_keys.append(('ln_vision.bias', 'vision_model.post_layernorm.bias')						)

		for i in range(config.vision_config.num_hidden_layers						):
				rename_keys.append((F"visual_encoder.blocks.{i}.norm1.weight", F"vision_model.encoder.layers.{i}.layer_norm1.weight")						)
				rename_keys.append((F"visual_encoder.blocks.{i}.norm1.bias", F"vision_model.encoder.layers.{i}.layer_norm1.bias")						)
				rename_keys.append((F"visual_encoder.blocks.{i}.norm2.weight", F"vision_model.encoder.layers.{i}.layer_norm2.weight")						)
				rename_keys.append((F"visual_encoder.blocks.{i}.norm2.bias", F"vision_model.encoder.layers.{i}.layer_norm2.bias")						)
				rename_keys.append((F"visual_encoder.blocks.{i}.attn.qkv.weight", F"vision_model.encoder.layers.{i}.self_attn.qkv.weight")						)
				rename_keys.append((F"visual_encoder.blocks.{i}.attn.proj.weight", F"vision_model.encoder.layers.{i}.self_attn.projection.weight",)						)
				rename_keys.append((F"visual_encoder.blocks.{i}.attn.proj.bias", F"vision_model.encoder.layers.{i}.self_attn.projection.bias")						)
				rename_keys.append((F"visual_encoder.blocks.{i}.mlp.fc1.weight", F"vision_model.encoder.layers.{i}.mlp.fc1.weight")						)
				rename_keys.append((F"visual_encoder.blocks.{i}.mlp.fc1.bias", F"vision_model.encoder.layers.{i}.mlp.fc1.bias")						)
				rename_keys.append((F"visual_encoder.blocks.{i}.mlp.fc2.weight", F"vision_model.encoder.layers.{i}.mlp.fc2.weight")						)
				rename_keys.append((F"visual_encoder.blocks.{i}.mlp.fc2.bias", F"vision_model.encoder.layers.{i}.mlp.fc2.bias")						)

		# QFormer
		rename_keys.append(('Qformer.bert.embeddings.LayerNorm.weight', 'qformer.layernorm.weight')						)
		rename_keys.append(('Qformer.bert.embeddings.LayerNorm.bias', 'qformer.layernorm.bias')						)

		# fmt: on
		return rename_keys


def 		_SCREAMING_SNAKE_CASE     ( lowerCAmelCase__							,lowerCAmelCase__							,lowerCAmelCase__						):
		lowerCamelCase_	:     Optional[int] 						=      dct.pop(lowerCAmelCase__						)
		lowerCamelCase_	:     List[Any] 						=      val


def 		_SCREAMING_SNAKE_CASE     ( lowerCAmelCase__							,lowerCAmelCase__						):
		for i in range(config.vision_config.num_hidden_layers						):
				# read in original q and v biases
				lowerCamelCase_	:     int 						=      state_dict.pop(F"visual_encoder.blocks.{i}.attn.q_bias"						)
				lowerCamelCase_	:     List[str] 						=      state_dict.pop(F"visual_encoder.blocks.{i}.attn.v_bias"						)

				# next, set bias in the state dict
				lowerCamelCase_	:     List[Any] 						=      torch.cat((q_bias, torch.zeros_like(lowerCAmelCase__							,requires_grad=lowerCAmelCase__						), v_bias)						)
				lowerCamelCase_	:     Optional[int] 						=      qkv_bias


def 		_SCREAMING_SNAKE_CASE     ( lowerCAmelCase__							,lowerCAmelCase__						):
		lowerCamelCase_	:     int 						=      3_64 if 'coco' in model_name else 2_24
		lowerCamelCase_	:     Any 						=      BlipaVisionConfig(image_size=lowerCAmelCase__						).to_dict()

		# make sure the models have proper bos_token_id and eos_token_id set (important for generation)
		# seems like flan-T5 models don't have bos_token_id properly set?
		if "opt-2.7b" in model_name:
				lowerCamelCase_	:     Optional[int] 						=      OPTConfig.from_pretrained('facebook/opt-2.7b'							,eos_token_id=lowerCAmelCase__						).to_dict()
		elif "opt-6.7b" in model_name:
				lowerCamelCase_	:     str 						=      OPTConfig.from_pretrained('facebook/opt-6.7b'							,eos_token_id=lowerCAmelCase__						).to_dict()
		elif "t5-xl" in model_name:
				lowerCamelCase_	:     Dict 						=      TaConfig.from_pretrained('google/flan-t5-xl'							,dense_act_fn='gelu'							,bos_token_id=1						).to_dict()
		elif "t5-xxl" in model_name:
				lowerCamelCase_	:     Any 						=      TaConfig.from_pretrained('google/flan-t5-xxl'							,dense_act_fn='gelu'							,bos_token_id=1						).to_dict()

		lowerCamelCase_	:     Any 						=      BlipaConfig(vision_config=lowerCAmelCase__							,text_config=lowerCAmelCase__						)

		return config, image_size


@torch.no_grad()
def 		_SCREAMING_SNAKE_CASE     ( lowerCAmelCase__							,lowerCAmelCase__=None							,lowerCAmelCase__=False						):
		lowerCamelCase_	:     List[str] 						=      (
		    AutoTokenizer.from_pretrained('facebook/opt-2.7b'						)
		    if 'opt' in model_name
		    else AutoTokenizer.from_pretrained('google/flan-t5-xl'						)
		)
		lowerCamelCase_	:     str 						=      tokenizer('\n'							,add_special_tokens=lowerCAmelCase__						).input_ids[0]
		lowerCamelCase_						,      lowerCamelCase_	:     Optional[int] 						=      get_blipa_config(lowerCAmelCase__							,eos_token_id=lowerCAmelCase__						)

		lowerCamelCase_	:     Optional[int] 						=      BlipaForConditionalGeneration(lowerCAmelCase__						).eval()

		lowerCamelCase_	:     Optional[Any] 						=      {
		    'blip2-opt-2.7b': ('blip2_opt', 'pretrain_opt2.7b'),
		    'blip2-opt-6.7b': ('blip2_opt', 'pretrain_opt6.7b'),
		    'blip2-opt-2.7b-coco': ('blip2_opt', 'caption_coco_opt2.7b'),
		    'blip2-opt-6.7b-coco': ('blip2_opt', 'caption_coco_opt6.7b'),
		    'blip2-flan-t5-xl': ('blip2_t5', 'pretrain_flant5xl'),
		    'blip2-flan-t5-xl-coco': ('blip2_t5', 'caption_coco_flant5xl'),
		    'blip2-flan-t5-xxl': ('blip2_t5', 'pretrain_flant5xxl'),
		}

		lowerCamelCase_						,      lowerCamelCase_	:     Optional[int] 						=      model_name_to_original[model_name]

		# load original model
		print('Loading original model...'						)
		lowerCamelCase_	:     Optional[Any] 						=      'cuda' if torch.cuda.is_available() else 'cpu'
		lowerCamelCase_						,      lowerCamelCase_						,      lowerCamelCase_	:     List[Any] 						=      load_model_and_preprocess(
		    name=lowerCAmelCase__							,model_type=lowerCAmelCase__							,is_eval=lowerCAmelCase__							,device=lowerCAmelCase__						)
		original_model.eval()
		print('Done!'						)

		# update state dict keys
		lowerCamelCase_	:     int 						=      original_model.state_dict()
		lowerCamelCase_	:     Dict 						=      create_rename_keys(lowerCAmelCase__						)
		for src, dest in rename_keys:
				rename_key(lowerCAmelCase__							,lowerCAmelCase__							,lowerCAmelCase__						)

		# some keys can be renamed efficiently
		for key, val in state_dict.copy().items():
				lowerCamelCase_	:     int 						=      state_dict.pop(lowerCAmelCase__						)
				if key.startswith('Qformer.bert'						):
						lowerCamelCase_	:     int 						=      key.replace('Qformer.bert'							,'qformer'						)
				if "attention.self" in key:
						lowerCamelCase_	:     Dict 						=      key.replace('self'							,'attention'						)
				if "opt_proj" in key:
						lowerCamelCase_	:     Optional[int] 						=      key.replace('opt_proj'							,'language_projection'						)
				if "t5_proj" in key:
						lowerCamelCase_	:     List[str] 						=      key.replace('t5_proj'							,'language_projection'						)
				if key.startswith('opt'						):
						lowerCamelCase_	:     Optional[int] 						=      key.replace('opt'							,'language'						)
				if key.startswith('t5'						):
						lowerCamelCase_	:     Optional[int] 						=      key.replace('t5'							,'language'						)
				lowerCamelCase_	:     Optional[Any] 						=      val

		# read in qv biases
		read_in_q_v_bias(lowerCAmelCase__							,lowerCAmelCase__						)

		lowerCamelCase_						,      lowerCamelCase_	:     Dict 						=      hf_model.load_state_dict(lowerCAmelCase__							,strict=lowerCAmelCase__						)
		assert len(lowerCAmelCase__						) == 0
		assert unexpected_keys == ["qformer.embeddings.position_ids"]

		lowerCamelCase_	:     Tuple 						=      load_demo_image()
		lowerCamelCase_	:     List[Any] 						=      vis_processors['eval'](lowerCAmelCase__						).unsqueeze(0						).to(lowerCAmelCase__						)
		lowerCamelCase_	:     Optional[Any] 						=      tokenizer(['\n']							,return_tensors='pt'						).input_ids.to(lowerCAmelCase__						)

		# create processor
		lowerCamelCase_	:     Union[str, Any] 						=      BlipImageProcessor(
		    size={'height': image_size, 'width': image_size}							,image_mean=lowerCAmelCase__							,image_std=lowerCAmelCase__						)
		lowerCamelCase_	:     Tuple 						=      BlipaProcessor(image_processor=lowerCAmelCase__							,tokenizer=lowerCAmelCase__						)
		lowerCamelCase_	:     Optional[int] 						=      processor(images=lowerCAmelCase__							,return_tensors='pt'						).pixel_values.to(lowerCAmelCase__						)

		# make sure processor creates exact same pixel values
		assert torch.allclose(lowerCAmelCase__							,lowerCAmelCase__						)

		original_model.to(lowerCAmelCase__						)
		hf_model.to(lowerCAmelCase__						)
		with torch.no_grad():
				if "opt" in model_name:
						lowerCamelCase_	:     List[Any] 						=      original_model({'image': original_pixel_values, 'text_input': ['']}						).logits
						lowerCamelCase_	:     Optional[int] 						=      hf_model(lowerCAmelCase__							,lowerCAmelCase__						).logits
				else:
						lowerCamelCase_	:     Optional[Any] 						=      original_model(
						    {'image': original_pixel_values, 'text_input': ['\n'], 'text_output': ['\n']}						).logits
						lowerCamelCase_	:     Dict 						=      input_ids.masked_fill(input_ids == tokenizer.pad_token_id							,-1_00						)
						lowerCamelCase_	:     Tuple 						=      hf_model(lowerCAmelCase__							,lowerCAmelCase__							,labels=lowerCAmelCase__						).logits

		assert original_logits.shape == logits.shape
		print('First values of original logits:'							,original_logits[0, :3, :3]						)
		print('First values of HF logits:'							,logits[0, :3, :3]						)

		# assert values
		if model_name == "blip2-flan-t5-xl":
				lowerCamelCase_	:     Union[str, Any] 						=      torch.tensor(
				    [[-41.5_850, -4.4_440, -8.9_922], [-47.4_322, -5.9_143, -1.7_340]]							,device=lowerCAmelCase__						)
				assert torch.allclose(logits[0, :3, :3]							,lowerCAmelCase__							,atol=1e-4						)
		elif model_name == "blip2-flan-t5-xl-coco":
				lowerCamelCase_	:     str 						=      torch.tensor(
				    [[-57.0_109, -9.8_967, -12.6_280], [-68.6_578, -12.7_191, -10.5_065]]							,device=lowerCAmelCase__						)
		else:
				# cast to same type
				lowerCamelCase_	:     str 						=      logits.dtype
				assert torch.allclose(original_logits.to(lowerCAmelCase__						)							,lowerCAmelCase__							,atol=1e-2						)
		print('Looks ok!'						)

		print('Generating a caption...'						)
		lowerCamelCase_	:     Optional[Any] 						=      ''
		lowerCamelCase_	:     Optional[Any] 						=      tokenizer(lowerCAmelCase__							,return_tensors='pt'						).input_ids.to(lowerCAmelCase__						)

		lowerCamelCase_	:     Optional[Any] 						=      original_model.generate({'image': original_pixel_values}						)
		lowerCamelCase_	:     Optional[Any] 						=      hf_model.generate(
		    lowerCAmelCase__							,lowerCAmelCase__							,do_sample=lowerCAmelCase__							,num_beams=5							,max_length=30							,min_length=1							,top_p=0.9							,repetition_penalty=1.0							,length_penalty=1.0							,temperature=1							,)
		print('Original generation:'							,lowerCAmelCase__						)
		lowerCamelCase_	:     List[Any] 						=      input_ids.shape[1]
		lowerCamelCase_	:     Optional[int] 						=      processor.batch_decode(outputs[:, prompt_length:]							,skip_special_tokens=lowerCAmelCase__						)
		lowerCamelCase_	:     Tuple 						=      [text.strip() for text in output_text]
		print('HF generation:'							,lowerCAmelCase__						)

		if pytorch_dump_folder_path is not None:
				processor.save_pretrained(lowerCAmelCase__						)
				hf_model.save_pretrained(lowerCAmelCase__						)

		if push_to_hub:
				processor.push_to_hub(F"nielsr/{model_name}"						)
				hf_model.push_to_hub(F"nielsr/{model_name}"						)


if __name__ == "__main__":
		_lowercase		:  Any			       =argparse.ArgumentParser()
		_lowercase		:  Union[str, Any]			       =[
		    """blip2-opt-2.7b""",
		    """blip2-opt-6.7b""",
		    """blip2-opt-2.7b-coco""",
		    """blip2-opt-6.7b-coco""",
		    """blip2-flan-t5-xl""",
		    """blip2-flan-t5-xl-coco""",
		    """blip2-flan-t5-xxl""",
		]
		parser.add_argument(
		    """--model_name""",
		    default="""blip2-opt-2.7b""",
		    choices=choices,
		    type=str,
		    help="""Path to hf config.json of model to convert""",
		)
		parser.add_argument("""--pytorch_dump_folder_path""", default=None, type=str, help="""Path to the output PyTorch model.""")
		parser.add_argument(
		    """--push_to_hub""",
		    action="""store_true""",
		    help="""Whether to push the model and processor to the hub after converting""",
		)

		_lowercase		:  int			       =parser.parse_args()

		convert_blipa_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)







import os
import sys

_lowercase		:  List[str]			       =os.path.join(os.path.dirname(__file__), """src""")
sys.path.append(SRC_DIR)


from transformers import (
    AutoConfig,
    AutoModel,
    AutoModelForCausalLM,
    AutoModelForMaskedLM,
    AutoModelForQuestionAnswering,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    add_start_docstrings,
)


_lowercase		:  Optional[int]			       =[
    """torch""",
    """numpy""",
    """tokenizers""",
    """filelock""",
    """requests""",
    """tqdm""",
    """regex""",
    """sentencepiece""",
    """sacremoses""",
    """importlib_metadata""",
    """huggingface_hub""",
]


@add_start_docstrings(AutoConfig.__doc__						)
def 		_SCREAMING_SNAKE_CASE     ( *lowerCAmelCase__							,**lowerCAmelCase__						):
		return AutoConfig.from_pretrained(*lowerCAmelCase__							,**lowerCAmelCase__						)


@add_start_docstrings(AutoTokenizer.__doc__						)
def 		_SCREAMING_SNAKE_CASE     ( *lowerCAmelCase__							,**lowerCAmelCase__						):
		return AutoTokenizer.from_pretrained(*lowerCAmelCase__							,**lowerCAmelCase__						)


@add_start_docstrings(AutoModel.__doc__						)
def 		_SCREAMING_SNAKE_CASE     ( *lowerCAmelCase__							,**lowerCAmelCase__						):
		return AutoModel.from_pretrained(*lowerCAmelCase__							,**lowerCAmelCase__						)


@add_start_docstrings(AutoModelForCausalLM.__doc__						)
def 		_SCREAMING_SNAKE_CASE     ( *lowerCAmelCase__							,**lowerCAmelCase__						):
		return AutoModelForCausalLM.from_pretrained(*lowerCAmelCase__							,**lowerCAmelCase__						)


@add_start_docstrings(AutoModelForMaskedLM.__doc__						)
def 		_SCREAMING_SNAKE_CASE     ( *lowerCAmelCase__							,**lowerCAmelCase__						):
		return AutoModelForMaskedLM.from_pretrained(*lowerCAmelCase__							,**lowerCAmelCase__						)


@add_start_docstrings(AutoModelForSequenceClassification.__doc__						)
def 		_SCREAMING_SNAKE_CASE     ( *lowerCAmelCase__							,**lowerCAmelCase__						):
		return AutoModelForSequenceClassification.from_pretrained(*lowerCAmelCase__							,**lowerCAmelCase__						)





@add_start_docstrings(AutoModelForQuestionAnswering.__doc__						)
def 		_SCREAMING_SNAKE_CASE     ( *lowerCAmelCase__							,**lowerCAmelCase__						):
		return AutoModelForQuestionAnswering.from_pretrained(*lowerCAmelCase__							,**lowerCAmelCase__						)
