


"""simple docstring"""







import argparse
import os
import re

import tensorflow as tf
import torch

from transformers import BertConfig, BertModel
from transformers.utils import logging


logging.set_verbosity_info()
a__       :							int        =  logging.get_logger(__name__)




def      UpperCAmelCase__ (lowerCAmelCase_						,		lowerCAmelCase_						,		lowerCAmelCase_     ):




					'''simple docstring'''


					__SCREAMING_SNAKE_CASE		     =       os.path.abspath(lowerCAmelCase_     )
					logger.info(f"""Converting TensorFlow checkpoint from {tf_path}"""     )
					# Load weights from TF model
					__SCREAMING_SNAKE_CASE		     =       tf.train.list_variables(lowerCAmelCase_     )
					__SCREAMING_SNAKE_CASE		     =       []
					__SCREAMING_SNAKE_CASE		     =       []
					__SCREAMING_SNAKE_CASE		     =       []
					for full_name, shape in init_vars:
										# logger.info(f"Loading TF weight {name} with shape {shape}")
										__SCREAMING_SNAKE_CASE		     =       full_name.split("/"     )
										if full_name == "_CHECKPOINTABLE_OBJECT_GRAPH" or name[0] in ["global_step", "save_counter"]:
															logger.info(f"""Skipping non-model layer {full_name}"""     )
															continue
										if "optimizer" in full_name:
															logger.info(f"""Skipping optimization layer {full_name}"""     )
															continue
										if name[0] == "model":
															# ignore initial 'model'
															__SCREAMING_SNAKE_CASE		     =       name[1:]
										# figure out how many levels deep the name is
										__SCREAMING_SNAKE_CASE		     =       0
										for _name in name:
															if _name.startswith("layer_with_weights"     ):
																				depth += 1
															else:
																				break
										layer_depth.append(lowerCAmelCase_     )
										# read data
										__SCREAMING_SNAKE_CASE		     =       tf.train.load_variable(lowerCAmelCase_						,		lowerCAmelCase_     )
										names.append("/".join(lowerCAmelCase_     )     )
										arrays.append(lowerCAmelCase_     )
					logger.info(f"""Read a total of {len(lowerCAmelCase_     ):,} layers"""     )

					# Sanity check
					if len(set(lowerCAmelCase_     )     ) != 1:
										raise ValueError(f"""Found layer names with different depths (layer depth {list(set(lowerCAmelCase_     )     )})"""     )
					__SCREAMING_SNAKE_CASE		     =       list(set(lowerCAmelCase_     )     )[0]
					if layer_depth != 1:
										raise ValueError(
										    "The model contains more than just the embedding/encoder layers. This script does not handle MLM/NSP"
										    " heads."     )

					# convert layers
					logger.info("Converting weights..."     )
					for full_name, array in zip(lowerCAmelCase_						,		lowerCAmelCase_     ):
										__SCREAMING_SNAKE_CASE		     =       full_name.split("/"     )
										__SCREAMING_SNAKE_CASE		     =       model
										__SCREAMING_SNAKE_CASE		     =       []
										for i, m_name in enumerate(lowerCAmelCase_     ):
															if m_name == ".ATTRIBUTES":
																				# variable names end with .ATTRIBUTES/VARIABLE_VALUE
																				break
															if m_name.startswith("layer_with_weights"     ):
																				__SCREAMING_SNAKE_CASE		     =       int(m_name.split("-"     )[-1]     )
																				if layer_num <= 2:
																									# embedding layers
																									# layer_num 0: word_embeddings
																									# layer_num 1: position_embeddings
																									# layer_num 2: token_type_embeddings
																									continue
																				elif layer_num == 3:
																									# embedding LayerNorm
																									trace.extend(["embeddings", "LayerNorm"]     )
																									__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"embeddings"     )
																									__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"LayerNorm"     )
																				elif layer_num > 3 and layer_num < config.num_hidden_layers + 4:
																									# encoder layers
																									trace.extend(["encoder", "layer", str(layer_num - 4     )]     )
																									__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"encoder"     )
																									__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"layer"     )
																									__SCREAMING_SNAKE_CASE		     =       pointer[layer_num - 4]
																				elif layer_num == config.num_hidden_layers + 4:
																									# pooler layer
																									trace.extend(["pooler", "dense"]     )
																									__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"pooler"     )
																									__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"dense"     )
															elif m_name == "embeddings":
																				trace.append("embeddings"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"embeddings"     )
																				if layer_num == 0:
																									trace.append("word_embeddings"     )
																									__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"word_embeddings"     )
																				elif layer_num == 1:
																									trace.append("position_embeddings"     )
																									__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"position_embeddings"     )
																				elif layer_num == 2:
																									trace.append("token_type_embeddings"     )
																									__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"token_type_embeddings"     )
																				else:
																									raise ValueError(f"""Unknown embedding layer with name {full_name}"""     )
																				trace.append("weight"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"weight"     )
															elif m_name == "_attention_layer":
																				# self-attention layer
																				trace.extend(["attention", "self"]     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"attention"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"self"     )
															elif m_name == "_attention_layer_norm":
																				# output attention norm
																				trace.extend(["attention", "output", "LayerNorm"]     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"attention"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"output"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"LayerNorm"     )
															elif m_name == "_attention_output_dense":
																				# output attention dense
																				trace.extend(["attention", "output", "dense"]     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"attention"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"output"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"dense"     )
															elif m_name == "_output_dense":
																				# output dense
																				trace.extend(["output", "dense"]     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"output"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"dense"     )
															elif m_name == "_output_layer_norm":
																				# output dense
																				trace.extend(["output", "LayerNorm"]     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"output"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"LayerNorm"     )
															elif m_name == "_key_dense":
																				# attention key
																				trace.append("key"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"key"     )
															elif m_name == "_query_dense":
																				# attention query
																				trace.append("query"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"query"     )
															elif m_name == "_value_dense":
																				# attention value
																				trace.append("value"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"value"     )
															elif m_name == "_intermediate_dense":
																				# attention intermediate dense
																				trace.extend(["intermediate", "dense"]     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"intermediate"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"dense"     )
															elif m_name == "_output_layer_norm":
																				# output layer norm
																				trace.append("output"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"output"     )
															# weights & biases
															elif m_name in ["bias", "beta"]:
																				trace.append("bias"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"bias"     )
															elif m_name in ["kernel", "gamma"]:
																				trace.append("weight"     )
																				__SCREAMING_SNAKE_CASE		     =       getattr(lowerCAmelCase_						,		"weight"     )
															else:
																				logger.warning(f"""Ignored {m_name}"""     )
        # for certain layers reshape is necessary
										__SCREAMING_SNAKE_CASE		     =       ".".join(lowerCAmelCase_     )
										if re.match(R"(\S+)\.attention\.self\.(key|value|query)\.(bias|weight)"						,		lowerCAmelCase_     ) or re.match(
										    R"(\S+)\.attention\.output\.dense\.weight"						,		lowerCAmelCase_     ):
															__SCREAMING_SNAKE_CASE		     =       array.reshape(pointer.data.shape     )
										if "kernel" in full_name:
															__SCREAMING_SNAKE_CASE		     =       array.transpose()
										if pointer.shape == array.shape:
															__SCREAMING_SNAKE_CASE		     =       torch.from_numpy(lowerCAmelCase_     )
										else:
															raise ValueError(
															    f"""Shape mismatch in layer {full_name}: Model expects shape {pointer.shape} but layer contains shape:"""
															    f""" {array.shape}"""     )
										logger.info(f"""Successfully set variable {full_name} to PyTorch layer {trace}"""     )
					return model




def      UpperCAmelCase__ (lowerCAmelCase_						,		lowerCAmelCase_						,		lowerCAmelCase_     ):




					'''simple docstring'''


					logger.info(f"""Loading model based on config from {config_path}..."""     )
					__SCREAMING_SNAKE_CASE		     =       BertConfig.from_json_file(lowerCAmelCase_     )
					__SCREAMING_SNAKE_CASE		     =       BertModel(lowerCAmelCase_     )

					# Load weights from checkpoint
					logger.info(f"""Loading weights from checkpoint {tf_checkpoint_path}..."""     )
					load_tfa_weights_in_bert(lowerCAmelCase_						,		lowerCAmelCase_						,		lowerCAmelCase_     )

					# Save pytorch-model
					logger.info(f"""Saving PyTorch model to {pytorch_dump_path}..."""     )
					torch.save(model.state_dict()						,		lowerCAmelCase_     )


if __name__ == "__main__":
		a__       :							Union[str, Any]        =  argparse.ArgumentParser()
		parser.add_argument(
		    '''--tf_checkpoint_path''', type=str, required=True, help='''Path to the TensorFlow 2.x checkpoint path.'''
		)
		parser.add_argument(
		    '''--bert_config_file''',
		    type=str,
		    required=True,
		    help='''The config json file corresponding to the BERT model. This specifies the model architecture.''',
		)
		parser.add_argument(
		    '''--pytorch_dump_path''',
		    type=str,
		    required=True,
		    help='''Path to the output PyTorch model (must include filename).''',
		)
		a__       :							Tuple        =  parser.parse_args()
		convert_tfa_checkpoint_to_pytorch(args.tf_checkpoint_path, args.bert_config_file, args.pytorch_dump_path)




"""simple docstring"""







import unittest

from transformers import MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING, is_vision_available
from transformers.pipelines import pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
		from PIL import Image
else:




		class 				UpperCamelCase_			:

			"""simple docstring"""





			@staticmethod
			def 						UpperCAmelCase_   (				*UpperCAmelCase__    :       List[Any]    ,    **UpperCAmelCase__    :       Optional[Any]     ) ->			Optional[int]:
								pass




@is_pipeline_test
@require_torch
@require_vision
class 				UpperCamelCase_			(				unittest.TestCase):

	"""simple docstring"""





	snake_case__       :						List[Any]					      =    MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING




	def 						UpperCAmelCase_   (				self    :       Dict    ,    UpperCAmelCase__    :       Tuple    ,    UpperCAmelCase__    :       List[Any]    ,    UpperCAmelCase__    :       Optional[int]     ) ->			str:
						__SCREAMING_SNAKE_CASE		     =       pipeline("visual-question-answering"    ,    model="hf-internal-testing/tiny-vilt-random-vqa"     )
						__SCREAMING_SNAKE_CASE		     =       [
						    {
						        "image": Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"     ),
						        "question": "How many cats are there?",
						    },
						    {
						        "image": "./tests/fixtures/tests_samples/COCO/000000039769.png",
						        "question": "How many cats are there?",
						    },
						]
						return vqa_pipeline, examples




	def 						UpperCAmelCase_   (				self    :       Dict    ,    UpperCAmelCase__    :       List[Any]    ,    UpperCAmelCase__    :       Any     ) ->			Any:
						__SCREAMING_SNAKE_CASE		     =       vqa_pipeline(UpperCAmelCase__    ,    top_k=1     )
						self.assertEqual(
						    UpperCAmelCase__    ,    [
						        [{"score": ANY(UpperCAmelCase__     ), "answer": ANY(UpperCAmelCase__     )}],
						        [{"score": ANY(UpperCAmelCase__     ), "answer": ANY(UpperCAmelCase__     )}],
						    ]    ,    )




	@require_torch
	def 						UpperCAmelCase_   (				self    :       Optional[Any]     ) ->			int:
						__SCREAMING_SNAKE_CASE		     =       pipeline("visual-question-answering"    ,    model="hf-internal-testing/tiny-vilt-random-vqa"     )
						__SCREAMING_SNAKE_CASE		     =       "./tests/fixtures/tests_samples/COCO/000000039769.png"
						__SCREAMING_SNAKE_CASE		     =       "How many cats are there?"

						__SCREAMING_SNAKE_CASE		     =       vqa_pipeline(image=UpperCAmelCase__    ,    question="How many cats are there?"    ,    top_k=2     )
						self.assertEqual(
						    UpperCAmelCase__    ,    [{"score": ANY(UpperCAmelCase__     ), "answer": ANY(UpperCAmelCase__     )}, {"score": ANY(UpperCAmelCase__     ), "answer": ANY(UpperCAmelCase__     )}]     )

						__SCREAMING_SNAKE_CASE		     =       vqa_pipeline({"image": image, "question": question}    ,    top_k=2     )
						self.assertEqual(
						    UpperCAmelCase__    ,    [{"score": ANY(UpperCAmelCase__     ), "answer": ANY(UpperCAmelCase__     )}, {"score": ANY(UpperCAmelCase__     ), "answer": ANY(UpperCAmelCase__     )}]     )




	@slow
	@require_torch
	def 						UpperCAmelCase_   (				self    :       Tuple     ) ->			Any:
						__SCREAMING_SNAKE_CASE		     =       pipeline("visual-question-answering"    ,    model="dandelin/vilt-b32-finetuned-vqa"     )
						__SCREAMING_SNAKE_CASE		     =       "./tests/fixtures/tests_samples/COCO/000000039769.png"
						__SCREAMING_SNAKE_CASE		     =       "How many cats are there?"

						__SCREAMING_SNAKE_CASE		     =       vqa_pipeline(image=UpperCAmelCase__    ,    question=UpperCAmelCase__    ,    top_k=2     )
						self.assertEqual(
						    nested_simplify(UpperCAmelCase__    ,    decimals=4     )    ,    [{"score": 0.8_799, "answer": "2"}, {"score": 0.296, "answer": "1"}]     )

						__SCREAMING_SNAKE_CASE		     =       vqa_pipeline({"image": image, "question": question}    ,    top_k=2     )
						self.assertEqual(
						    nested_simplify(UpperCAmelCase__    ,    decimals=4     )    ,    [{"score": 0.8_799, "answer": "2"}, {"score": 0.296, "answer": "1"}]     )

						__SCREAMING_SNAKE_CASE		     =       vqa_pipeline(
						    [{"image": image, "question": question}, {"image": image, "question": question}]    ,    top_k=2     )
						self.assertEqual(
						    nested_simplify(UpperCAmelCase__    ,    decimals=4     )    ,    [[{"score": 0.8_799, "answer": "2"}, {"score": 0.296, "answer": "1"}]] * 2    ,    )




	@require_tf
	@unittest.skip("Visual question answering not implemented in TF"     )
	def 						UpperCAmelCase_   (				self    :       str     ) ->			Optional[Any]:
						pass
