





'''simple docstring'''





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       A__(		):
				_snake_case :			Optional[Any]  						=   """https://storage.googleapis.com/sfr-vision-language-research/LAVIS/assets/merlion.png"""
				_snake_case :			Union[str, Any]  						=   Image.open(requests.get(snake_case__	,			stream=snake_case__ ).raw ).convert('RGB' )

				return image

def       A__(		__lowerCAmelCase ):
				_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.layernorm.weight') )
				rename_keys.append(('Qformer.bert.embeddings.LayerNorm.bias', 'qformer.layernorm.bias') )

				# fmt: on
				return rename_keys

def       A__(		__lowerCAmelCase	,			__lowerCAmelCase	,			__lowerCAmelCase ):
				_snake_case :			Optional[Any]  						=   dct.pop(snake_case__ )
				_snake_case :			Optional[int]  						=   val

def       A__(		__lowerCAmelCase	,			__lowerCAmelCase ):

				for i in range(config.vision_config.num_hidden_layers ):
								# read in original q and v biases
								_snake_case :			Optional[int]  						=   state_dict.pop(F'''visual_encoder.blocks.{i}.attn.q_bias''' )
								_snake_case :			Tuple  						=   state_dict.pop(F'''visual_encoder.blocks.{i}.attn.v_bias''' )

								# next, set bias in the state dict
								_snake_case :			List[str]  						=   torch.cat((q_bias, torch.zeros_like(snake_case__	,			requires_grad=snake_case__ ), v_bias) )
								_snake_case :			Dict  						=   qkv_bias

def       A__(		__lowerCAmelCase	,			__lowerCAmelCase ):
				_snake_case :			List[Any]  						=   3_64 if """coco""" in model_name else 2_24
				_snake_case :			List[str]  						=   BlipaVisionConfig(image_size=snake_case__ ).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:
								_snake_case :			List[str]  						=   OPTConfig.from_pretrained('facebook/opt-2.7b'	,			eos_token_id=snake_case__ ).to_dict()
				elif "opt-6.7b" in model_name:
								_snake_case :			List[str]  						=   OPTConfig.from_pretrained('facebook/opt-6.7b'	,			eos_token_id=snake_case__ ).to_dict()
				elif "t5-xl" in model_name:
								_snake_case :			Tuple  						=   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 :			List[Any]  						=   TaConfig.from_pretrained('google/flan-t5-xxl'	,			dense_act_fn='gelu'	,			bos_token_id=1 ).to_dict()

				_snake_case :			int  						=   BlipaConfig(vision_config=snake_case__	,			text_config=snake_case__ )

				return config, image_size

@torch.no_grad()
def       A__(		__lowerCAmelCase	,			__lowerCAmelCase=None	,			__lowerCAmelCase=False ):
				_snake_case :			List[str]  						=   (
				    AutoTokenizer.from_pretrained('facebook/opt-2.7b' )
				    if """opt""" in model_name
				    else AutoTokenizer.from_pretrained('google/flan-t5-xl' )
				)
				_snake_case :			str  						=   tokenizer('\n'	,			add_special_tokens=snake_case__ ).input_ids[0]
				_snake_case :			Dict  						=   get_blipa_config(snake_case__	,			eos_token_id=snake_case__ )

				_snake_case :			str  						=   BlipaForConditionalGeneration(snake_case__ ).eval()

				_snake_case :			int  						=   {
				    """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"""),
				}

				_snake_case :			List[Any]  						=   model_name_to_original[model_name]

				# load original model
				print('Loading original model...' )
				_snake_case :			int  						=   """cuda""" if torch.cuda.is_available() else """cpu"""
				_snake_case :			Any  						=   load_model_and_preprocess(
				    name=snake_case__	,			model_type=snake_case__	,			is_eval=snake_case__	,			device=snake_case__ )
				original_model.eval()
				print('Done!' )

				# update state dict keys
				_snake_case :			Any  						=   original_model.state_dict()
				_snake_case :			Dict  						=   create_rename_keys(snake_case__ )
				for src, dest in rename_keys:
								rename_key(snake_case__	,			snake_case__	,			snake_case__ )

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

				# read in qv biases
				read_in_q_v_bias(snake_case__	,			snake_case__ )

				_snake_case :			List[str]  						=   hf_model.load_state_dict(snake_case__	,			strict=snake_case__ )
				assert len(snake_case__ ) == 0
				assert unexpected_keys == ["qformer.embeddings.position_ids"]

				_snake_case :			Any  						=   load_demo_image()
				_snake_case :			str  						=   vis_processors["""eval"""](snake_case__ ).unsqueeze(0 ).to(snake_case__ )
				_snake_case :			List[Any]  						=   tokenizer(['\n']	,			return_tensors='pt' ).input_ids.to(snake_case__ )

				# create processor
				_snake_case :			Any  						=   BlipImageProcessor(
				    size={'height': image_size, 'width': image_size}	,			image_mean=snake_case__	,			image_std=snake_case__ )
				_snake_case :			int  						=   BlipaProcessor(image_processor=snake_case__	,			tokenizer=snake_case__ )
				_snake_case :			Any  						=   processor(images=snake_case__	,			return_tensors='pt' ).pixel_values.to(snake_case__ )

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

				original_model.to(snake_case__ )
				hf_model.to(snake_case__ )
				with torch.no_grad():
								if "opt" in model_name:
												_snake_case :			str  						=   original_model({'image': original_pixel_values, 'text_input': ['']} ).logits
												_snake_case :			int  						=   hf_model(snake_case__	,			snake_case__ ).logits
								else:
												_snake_case :			str  						=   original_model(
												    {'image': original_pixel_values, 'text_input': ['\n'], 'text_output': ['\n']} ).logits
												_snake_case :			Optional[int]  						=   input_ids.masked_fill(input_ids == tokenizer.pad_token_id	,			-1_00 )
												_snake_case :			Union[str, Any]  						=   hf_model(snake_case__	,			snake_case__	,			labels=snake_case__ ).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":
								_snake_case :			List[str]  						=   torch.tensor(
								    [[-41.5_850, -4.4_440, -8.9_922], [-47.4_322, -5.9_143, -1.7_340]]	,			device=snake_case__ )
								assert torch.allclose(logits[0, :3, :3]	,			snake_case__	,			atol=1E-4 )
				elif model_name == "blip2-flan-t5-xl-coco":
								_snake_case :			Union[str, Any]  						=   torch.tensor(
								    [[-57.0_109, -9.8_967, -12.6_280], [-68.6_578, -12.7_191, -10.5_065]]	,			device=snake_case__ )
				else:
								# cast to same type
								_snake_case :			int  						=   logits.dtype
								assert torch.allclose(original_logits.to(snake_case__ )	,			snake_case__	,			atol=1E-2 )
				print('Looks ok!' )

				print('Generating a caption...' )
				_snake_case :			Any  						=   """"""
				_snake_case :			str  						=   tokenizer(snake_case__	,			return_tensors='pt' ).input_ids.to(snake_case__ )

				_snake_case :			Union[str, Any]  						=   original_model.generate({'image': original_pixel_values} )
				_snake_case :			Tuple  						=   hf_model.generate(
				    snake_case__	,			snake_case__	,			do_sample=snake_case__	,			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:'	,			snake_case__ )
				_snake_case :			Optional[Any]  						=   input_ids.shape[1]
				_snake_case :			int  						=   processor.batch_decode(outputs[:, prompt_length:]	,			skip_special_tokens=snake_case__ )
				_snake_case :			Optional[Any]  						=   [text.strip() for text in output_text]
				print('HF generation:'	,			snake_case__ )

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

				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_		:				List[Any]				=						argparse.ArgumentParser()
		lowercase_		:				str				=						[
		    '''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_		:				Optional[Any]				=						parser.parse_args()

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













from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


lowercase_		:				Optional[Any]				=						{
    '''configuration_roc_bert''': ['''ROC_BERT_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''RoCBertConfig'''],
    '''tokenization_roc_bert''': ['''RoCBertTokenizer'''],
}

try:
		if not is_tokenizers_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		pass

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		lowercase_		:				Tuple				=						[
		    '''ROC_BERT_PRETRAINED_MODEL_ARCHIVE_LIST''',
		    '''RoCBertForCausalLM''',
		    '''RoCBertForMaskedLM''',
		    '''RoCBertForMultipleChoice''',
		    '''RoCBertForPreTraining''',
		    '''RoCBertForQuestionAnswering''',
		    '''RoCBertForSequenceClassification''',
		    '''RoCBertForTokenClassification''',
		    '''RoCBertLayer''',
		    '''RoCBertModel''',
		    '''RoCBertPreTrainedModel''',
		    '''load_tf_weights_in_roc_bert''',
		]

if TYPE_CHECKING:
		from .configuration_roc_bert import ROC_BERT_PRETRAINED_CONFIG_ARCHIVE_MAP, RoCBertConfig
		from .tokenization_roc_bert import RoCBertTokenizer

		try:
				if not is_tokenizers_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				raise OptionalDependencyNotAvailable()

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_roc_bert import (
				    ROC_BERT_PRETRAINED_MODEL_ARCHIVE_LIST,
				    RoCBertForCausalLM,
				    RoCBertForMaskedLM,
				    RoCBertForMultipleChoice,
				    RoCBertForPreTraining,
				    RoCBertForQuestionAnswering,
				    RoCBertForSequenceClassification,
				    RoCBertForTokenClassification,
				    RoCBertLayer,
				    RoCBertModel,
				    RoCBertPreTrainedModel,
				    load_tf_weights_in_roc_bert,
				)


else:
		import sys

		lowercase_		:				Tuple				=						_LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)






