






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_float32 (there's also the fix_lavis branch)
# also note: to convert Vicuna checkpoints, we had to include /home/niels/python_projects/checkpoints/FastChat/vicuna-7b in lavis/configs/models/blip2/blip2_instruct_vicuna7b.yaml
# same for Vicuna-13b
from lavis.models import load_model_and_preprocess
from PIL import Image

from transformers import (
    AutoTokenizer,
    BlipImageProcessor,
    InstructBlipConfig,
    InstructBlipForConditionalGeneration,
    InstructBlipProcessor,
    InstructBlipQFormerConfig,
    InstructBlipVisionConfig,
    LlamaConfig,
    LlamaTokenizerFast,
    TaConfig,
    TaTokenizerFast,
)
from transformers.utils.constants import OPENAI_CLIP_MEAN, OPENAI_CLIP_STD







def 			A__							(      ):





 '''simple docstring'''







 snake_case__  :    List[str]							       ="""https://raw.githubusercontent.com/salesforce/LAVIS/main/docs/_static/Confusing-Pictures.jpg"""
 snake_case__  :    List[Any]							       =Image.open(requests.get(_a		,  stream=_a	).raw	).convert("""RGB"""	)

 return image







def 			A__							(      _a  :					Any	):





 '''simple docstring'''







 snake_case__  :    str							       =[]
 # 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.embeddings.layernorm.weight""")	)
 rename_keys.append(("""Qformer.bert.embeddings.LayerNorm.bias""", """qformer.embeddings.layernorm.bias""")	)

 # fmt: on
 return rename_keys







def 			A__							(      _a  :					Optional[Any]		,  _a  :					Optional[int]		,  _a  :					Optional[Any]	):





 '''simple docstring'''







 snake_case__  :    Optional[int]							       =dct.pop(_a	)
 snake_case__  :    Union[str, Any]							       =val







def 			A__							(      _a  :					Optional[int]		,  _a  :					List[Any]	):





 '''simple docstring'''







 for i in range(config.vision_config.num_hidden_layers	):
  # read in original q and v biases
  snake_case__  :    Dict							       =state_dict.pop(f"visual_encoder.blocks.{i}.attn.q_bias"	)
  snake_case__  :    List[Any]							       =state_dict.pop(f"visual_encoder.blocks.{i}.attn.v_bias"	)

  # next, set bias in the state dict
  snake_case__  :    Tuple							       =torch.cat((q_bias, torch.zeros_like(_a		,  requires_grad=_a	), v_bias)	)
  snake_case__  :    Union[str, Any]							       =qkv_bias







def 			A__							(      _a  :					int	):





 '''simple docstring'''







 snake_case__  :    List[Any]							       =364 if """coco""" in model_name else 224
 snake_case__  :    str							       =InstructBlipVisionConfig(image_size=_a	).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 "t5-xl" in model_name:
  snake_case__  :    Union[str, Any]							       =TaConfig.from_pretrained("""google/flan-t5-xl"""		,  dense_act_fn="""gelu"""		,  bos_token_id=1	).to_dict()
 elif "t5-xxl" in model_name:
  snake_case__  :    Any							       =TaConfig.from_pretrained("""google/flan-t5-xxl"""		,  dense_act_fn="""gelu"""		,  bos_token_id=1	).to_dict()
 elif "vicuna-7b" in model_name:
  snake_case__  :    List[Any]							       =LlamaConfig.from_pretrained("""decapoda-research/llama-7b-hf"""		,  vocab_size=32001	).to_dict()
 elif "vicuna-13b" in model_name:
  snake_case__  :    Union[str, Any]							       =LlamaConfig.from_pretrained("""decapoda-research/llama-13b-hf"""		,  vocab_size=32001	).to_dict()
 else:
  raise ValueError("""Model name not supported"""	)

 # the authors add one special "[DEC]" token to the vocab of Q-Former, hence vocab size = 30522 + 1
 snake_case__  :    Any							       =InstructBlipQFormerConfig(vocab_size=30523	).to_dict()
 snake_case__  :    Union[str, Any]							       =InstructBlipConfig(vision_config=_a		,  text_config=_a		,  qformer_config=_a	)

 return config, image_size







@torch.no_grad()
def 			A__							(      _a  :					List[Any]		,  _a  :					str=None		,  _a  :					Optional[int]=False	):





 '''simple docstring'''







 snake_case__  :    Union[str, Any]							       =AutoTokenizer.from_pretrained("""bert-base-uncased"""		,  truncation_side="""left"""	)
 qformer_tokenizer.add_special_tokens({"""bos_token""": """[DEC]"""}	)

 if "t5" in model_name:
  snake_case__  :    Tuple							       =TaTokenizerFast.from_pretrained("""google/flan-t5-xl"""		,  truncation_side="""left"""	)
 elif "vicuna" in model_name:
  # the following was used in the original implementation:
  # tokenizer = LlamaTokenizer.from_pretrained("huggyllama/llama-7b", use_fast=False, truncation_side="left")
  # tokenizer.add_special_tokens({"pad_token": "[PAD]"})
  # tokenizer.add_special_tokens({"bos_token": "</s>"})
  # tokenizer.add_special_tokens({"eos_token": "</s>"})
  # tokenizer.add_special_tokens({"unk_token": "</s>"})
  snake_case__  :    Union[str, Any]							       =LlamaTokenizerFast.from_pretrained(
      """huggyllama/llama-7b"""		,  truncation_side="""left"""		,  bos_token="""</s>"""		,  unk_token="""</s>"""	)
  tokenizer.add_special_tokens({"""pad_token""": """[PAD]"""}	)

 snake_case__       ,   snake_case__  :    int							       =get_blipa_config(_a	)
 snake_case__  :    Dict							       =InstructBlipForConditionalGeneration(_a	).eval()

 snake_case__  :    Union[str, Any]							       ={
     """instructblip-vicuna-7b""": ("""blip2_vicuna_instruct""", """vicuna7b"""),
     """instructblip-vicuna-13b""": ("""blip2_vicuna_instruct""", """vicuna13b"""),
     """instructblip-flan-t5-xl""": ("""blip2_t5_instruct""", """flant5xl"""),
     """instructblip-flan-t5-xxl""": ("""blip2_t5_instruct""", """flant5xxl"""),
 }

 snake_case__       ,   snake_case__  :    int							       =model_name_to_original[model_name]

 # load original model
 print("""Loading original model..."""	)
 snake_case__  :    Optional[Any]							       ="""cuda:1""" if torch.cuda.is_available() else """cpu"""
 snake_case__  :    Optional[int]							       ="""cuda:2""" if torch.cuda.is_available() else """cpu"""
 snake_case__       ,   snake_case__       ,   snake_case__  :    Union[str, Any]							       =load_model_and_preprocess(
     name=_a		,  model_type=_a		,  is_eval=_a		,  device=_a	)
 original_model.eval()
 print("""Done!"""	)

 # update state dict keys
 snake_case__  :    Optional[int]							       =original_model.state_dict()
 snake_case__  :    Any							       =create_rename_keys(_a	)
 for src, dest in rename_keys:
  rename_key(_a		,  _a		,  _a	)

 # some keys can be renamed efficiently
 for key, val in state_dict.copy().items():
  snake_case__  :    str							       =state_dict.pop(_a	)
  if key.startswith("""Qformer.bert"""	):
   snake_case__  :    List[str]							       =key.replace("""Qformer.bert"""		,  """qformer"""	)
  if "attention.self" in key:
   snake_case__  :    List[str]							       =key.replace("""self"""		,  """attention"""	)
  if "llm_proj" in key:
   snake_case__  :    List[str]							       =key.replace("""llm_proj"""		,  """language_projection"""	)
  if "t5_proj" in key:
   snake_case__  :    Optional[int]							       =key.replace("""t5_proj"""		,  """language_projection"""	)
  if key.startswith("""llm_model"""	):
   snake_case__  :    List[Any]							       =key.replace("""llm_model"""		,  """language_model"""	)
  if key.startswith("""t5"""	):
   snake_case__  :    List[Any]							       =key.replace("""t5"""		,  """language"""	)
  snake_case__  :    List[str]							       =val

 # read in qv biases
 read_in_q_v_bias(_a		,  _a	)

 # note: weights get loaded in torch.float32 by default
 hf_model.load_state_dict(_a		,  strict=_a	)

 snake_case__  :    str							       =load_demo_image()
 snake_case__  :    Dict							       ="""What is unusual about this image?"""

 # create processor
 snake_case__  :    Tuple							       =BlipImageProcessor(
     size={"""height""": image_size, """width""": image_size}		,  image_mean=_a		,  image_std=_a	)
 snake_case__  :    Union[str, Any]							       =InstructBlipProcessor(
     image_processor=_a		,  tokenizer=_a		,  qformer_tokenizer=_a		,  )
 snake_case__  :    int							       =processor(images=_a		,  text=_a		,  return_tensors="""pt"""	).to(_a	)

 # make sure processor creates exact same pixel values
 snake_case__  :    Tuple							       =vis_processors["""eval"""](_a	).unsqueeze(0	).to(_a	)
 snake_case__  :    str							       =inputs.pixel_values
 assert torch.allclose(original_pixel_values.to(pixel_values.device	)		,  _a	)

 original_model.to(_a	)
 hf_model.to(_a	)
 with torch.no_grad():
  if "vicuna" in model_name:
   snake_case__  :    str							       =original_model({"""image""": original_pixel_values, """text_input""": [prompt]}	).logits
   snake_case__  :    Tuple							       =hf_model(**_a	).logits
  else:
   snake_case__  :    Optional[int]							       =original_model(
       {"""image""": original_pixel_values, """text_input""": [prompt], """text_output""": ["""\n"""]}	).logits
   snake_case__  :    Optional[int]							       =tokenizer("""\n"""		,  return_tensors="""pt"""	).input_ids.to(_a	)
   snake_case__  :    Dict							       =label_input_ids.masked_fill(label_input_ids == tokenizer.pad_token_id		,  -100	)
   snake_case__  :    Dict							       =hf_model(**_a		,  labels=_a	).logits

 print("""First values of original logits:"""		,  original_logits[0, :3, :3]	)
 print("""First values of HF logits:"""		,  logits[0, :3, :3]	)

 # assert values
 assert original_logits.shape == logits.shape
 snake_case__  :    List[str]							       =1E-4 if """vicuna""" in model_name else 1E-5
 assert torch.allclose(original_logits.to(logits.device	)		,  _a		,  atol=_a	)
 print("""Looks ok!"""	)

 print("""Generating with original model..."""	)
 snake_case__  :    str							       =original_model.generate({"""image""": original_pixel_values, """prompt""": prompt}		,  num_beams=5	)

 # important: we need to cast the weights of the HF model to the appropriate type
 print("""Generating with HF model..."""	)
 snake_case__  :    Any							       =hf_model.generate(
     **_a		,  do_sample=_a		,  num_beams=5		,  max_length=256		,  min_length=1		,  top_p=0.9		,  repetition_penalty=1.5		,  length_penalty=1.0		,  temperature=1		,  )
 if "vicuna" in model_name:
  # convert output id 0 to 2 (eos_token_id)
  # TODO add this in the generate method?
  snake_case__  :    Any							       =2
 print("""Original generation:"""		,  _a	)
 snake_case__  :    List[Any]							       =processor.batch_decode(_a		,  skip_special_tokens=_a	)
 snake_case__  :    Dict							       =[text.strip() for text in output_text]
 print("""HF generation:"""		,  _a	)

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

 if push_to_hub:
  processor.push_to_hub(f"Salesforce/{model_name}"	)
  hf_model.push_to_hub(f"Salesforce/{model_name}"	)


if __name__ == "__main__":
       __lowerCamelCase	:		Dict										=   argparse.ArgumentParser()
       __lowerCamelCase	:		List[Any]										=   [
           """instructblip-vicuna-7b""",
           """instructblip-vicuna-13b""",
           """instructblip-flan-t5-xl""",
           """instructblip-flan-t5-xxl""",
       ]
       parser.add_argument(
           """--model_name""",
           default="""instructblip-flan-t5-xl""",
           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""",
       )

       __lowerCamelCase	:		Union[str, Any]										=   parser.parse_args()

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













def 			A__							(      _a  :					Optional[Any]		,  _a  :					Tuple		,  _a  :					List[str]=False	):





 '''simple docstring'''







 if isinstance(_a		,  _a	) and isinstance(_a		,  _a	):
  snake_case__  :    int							       =len(set_a.intersection(_a	)	)

  if alternative_union:
   snake_case__  :    int							       =len(_a	) + len(_a	)
  else:
   snake_case__  :    Tuple							       =len(set_a.union(_a	)	)

  return intersection / union

 if isinstance(_a		,  (list, tuple)	) and isinstance(_a		,  (list, tuple)	):
  snake_case__  :    Optional[int]							       =[element for element in set_a if element in set_b]

  if alternative_union:
   snake_case__  :    Optional[Any]							       =len(_a	) + len(_a	)
   return len(_a	) / union
  else:
   snake_case__  :    Dict							       =set_a + [element for element in set_b if element not in set_a]
   return len(_a	) / len(_a	)

  return len(_a	) / len(_a	)
 return None


if __name__ == "__main__":
       __lowerCamelCase	:		List[Any]										=   {"""a""", """b""", """c""", """d""", """e"""}
       __lowerCamelCase	:		Optional[int]										=   {"""c""", """d""", """e""", """f""", """h""", """i"""}
       print(jaccard_similarity(set_a, set_b))





