

import argparse
import os

import align
import numpy as np
import requests
import tensorflow as tf
import torch
from PIL import Image
from tokenizer import Tokenizer

from transformers import (
    AlignConfig,
    AlignModel,
    AlignProcessor,
    BertConfig,
    BertTokenizer,
    EfficientNetConfig,
    EfficientNetImageProcessor,
)
from transformers.utils import logging


logging.set_verbosity_info()
__A					=	logging.get_logger(__name__)



def       snake_case_(_UpperCamelCase     )							->	Union[str, Any]:



  """simple docstring"""






  _snake_case						=							tf.image.resize(_SCREAMING_SNAKE_CASE ,       (346, 346)     )
  _snake_case						=							tf.image.crop_to_bounding_box(_SCREAMING_SNAKE_CASE ,       (346 - 289) // 2 ,       (346 - 289) // 2 ,       289 ,       289     )
  return image



def       snake_case_()							->	Dict:



  """simple docstring"""






  _snake_case						=							EfficientNetConfig.from_pretrained('''google/efficientnet-b7'''     )
  _snake_case						=							289
  _snake_case						=							640
  _snake_case						=							{"0": "LABEL_0", "1": "LABEL_1"}
  _snake_case						=							{"LABEL_0": 0, "LABEL_1": 1}
  _snake_case						=							[]

  _snake_case						=							BertConfig()
  _snake_case						=							AlignConfig.from_text_vision_configs(
      text_config=_SCREAMING_SNAKE_CASE ,       vision_config=_SCREAMING_SNAKE_CASE ,       projection_dim=640     )
  return config



def       snake_case_()							->	Dict:



  """simple docstring"""






  _snake_case						=							"http://images.cocodataset.org/val2017/000000039769.jpg"
  _snake_case						=							Image.open(requests.get(_SCREAMING_SNAKE_CASE ,       stream=_SCREAMING_SNAKE_CASE     ).raw     )
  return im



def       snake_case_()							->	Tuple:



  """simple docstring"""






  _snake_case						=							EfficientNetImageProcessor(
      do_center_crop=_SCREAMING_SNAKE_CASE ,       rescale_factor=1 / 127.5 ,       rescale_offset=_SCREAMING_SNAKE_CASE ,       do_normalize=_SCREAMING_SNAKE_CASE ,       include_top=_SCREAMING_SNAKE_CASE ,       resample=Image.BILINEAR ,       )
  _snake_case						=							BertTokenizer.from_pretrained('''bert-base-uncased'''     )
  _snake_case						=							64
  _snake_case						=							AlignProcessor(image_processor=_SCREAMING_SNAKE_CASE ,       tokenizer=_SCREAMING_SNAKE_CASE     )
  return processor



def       snake_case_(_UpperCamelCase     )							->	Dict:



  """simple docstring"""






  _snake_case						=							[v.split('''_'''     )[0].split('''block'''     )[1] for v in original_param_names if v.startswith('''block'''     )]
  _snake_case						=							list(set(_SCREAMING_SNAKE_CASE     )     )
  _snake_case						=							sorted(_SCREAMING_SNAKE_CASE     )
  _snake_case						=							len(_SCREAMING_SNAKE_CASE     )
  _snake_case						=							{b: str(_SCREAMING_SNAKE_CASE     ) for b, i in zip(_SCREAMING_SNAKE_CASE ,       range(_SCREAMING_SNAKE_CASE     )     )}

  _snake_case						=							[]
  rename_keys.append(('''stem_conv/kernel:0''', '''embeddings.convolution.weight''')     )
  rename_keys.append(('''stem_bn/gamma:0''', '''embeddings.batchnorm.weight''')     )
  rename_keys.append(('''stem_bn/beta:0''', '''embeddings.batchnorm.bias''')     )
  rename_keys.append(('''stem_bn/moving_mean:0''', '''embeddings.batchnorm.running_mean''')     )
  rename_keys.append(('''stem_bn/moving_variance:0''', '''embeddings.batchnorm.running_var''')     )

  for b in block_names:
    _snake_case						=							block_name_mapping[b]
    rename_keys.append((F"""block{b}_expand_conv/kernel:0""", F"""encoder.blocks.{hf_b}.expansion.expand_conv.weight""")     )
    rename_keys.append((F"""block{b}_expand_bn/gamma:0""", F"""encoder.blocks.{hf_b}.expansion.expand_bn.weight""")     )
    rename_keys.append((F"""block{b}_expand_bn/beta:0""", F"""encoder.blocks.{hf_b}.expansion.expand_bn.bias""")     )
    rename_keys.append(
        (F"""block{b}_expand_bn/moving_mean:0""", F"""encoder.blocks.{hf_b}.expansion.expand_bn.running_mean""")     )
    rename_keys.append(
        (F"""block{b}_expand_bn/moving_variance:0""", F"""encoder.blocks.{hf_b}.expansion.expand_bn.running_var""")     )
    rename_keys.append(
        (F"""block{b}_dwconv/depthwise_kernel:0""", F"""encoder.blocks.{hf_b}.depthwise_conv.depthwise_conv.weight""")     )
    rename_keys.append((F"""block{b}_bn/gamma:0""", F"""encoder.blocks.{hf_b}.depthwise_conv.depthwise_norm.weight""")     )
    rename_keys.append((F"""block{b}_bn/beta:0""", F"""encoder.blocks.{hf_b}.depthwise_conv.depthwise_norm.bias""")     )
    rename_keys.append(
        (F"""block{b}_bn/moving_mean:0""", F"""encoder.blocks.{hf_b}.depthwise_conv.depthwise_norm.running_mean""")     )
    rename_keys.append(
        (F"""block{b}_bn/moving_variance:0""", F"""encoder.blocks.{hf_b}.depthwise_conv.depthwise_norm.running_var""")     )

    rename_keys.append((F"""block{b}_se_reduce/kernel:0""", F"""encoder.blocks.{hf_b}.squeeze_excite.reduce.weight""")     )
    rename_keys.append((F"""block{b}_se_reduce/bias:0""", F"""encoder.blocks.{hf_b}.squeeze_excite.reduce.bias""")     )
    rename_keys.append((F"""block{b}_se_expand/kernel:0""", F"""encoder.blocks.{hf_b}.squeeze_excite.expand.weight""")     )
    rename_keys.append((F"""block{b}_se_expand/bias:0""", F"""encoder.blocks.{hf_b}.squeeze_excite.expand.bias""")     )
    rename_keys.append(
        (F"""block{b}_project_conv/kernel:0""", F"""encoder.blocks.{hf_b}.projection.project_conv.weight""")     )
    rename_keys.append((F"""block{b}_project_bn/gamma:0""", F"""encoder.blocks.{hf_b}.projection.project_bn.weight""")     )
    rename_keys.append((F"""block{b}_project_bn/beta:0""", F"""encoder.blocks.{hf_b}.projection.project_bn.bias""")     )
    rename_keys.append(
        (F"""block{b}_project_bn/moving_mean:0""", F"""encoder.blocks.{hf_b}.projection.project_bn.running_mean""")     )
    rename_keys.append(
        (F"""block{b}_project_bn/moving_variance:0""", F"""encoder.blocks.{hf_b}.projection.project_bn.running_var""")     )

  _snake_case						=							{}
  for item in rename_keys:
    if item[0] in original_param_names:
      _snake_case						=							"vision_model." + item[1]

    # BERT text encoder
  _snake_case						=							[]
  _snake_case						=							"tf_bert_model/bert"
  _snake_case						=							"text_model"
  for i in range(12     ):
    rename_keys.append(
        (
            F"""{old}/encoder/layer_._{i}/attention/self/query/kernel:0""",
            F"""{new}.encoder.layer.{i}.attention.self.query.weight""",
        )     )
    rename_keys.append(
        (
            F"""{old}/encoder/layer_._{i}/attention/self/query/bias:0""",
            F"""{new}.encoder.layer.{i}.attention.self.query.bias""",
        )     )
    rename_keys.append(
        (
            F"""{old}/encoder/layer_._{i}/attention/self/key/kernel:0""",
            F"""{new}.encoder.layer.{i}.attention.self.key.weight""",
        )     )
    rename_keys.append(
        (
            F"""{old}/encoder/layer_._{i}/attention/self/key/bias:0""",
            F"""{new}.encoder.layer.{i}.attention.self.key.bias""",
        )     )
    rename_keys.append(
        (
            F"""{old}/encoder/layer_._{i}/attention/self/value/kernel:0""",
            F"""{new}.encoder.layer.{i}.attention.self.value.weight""",
        )     )
    rename_keys.append(
        (
            F"""{old}/encoder/layer_._{i}/attention/self/value/bias:0""",
            F"""{new}.encoder.layer.{i}.attention.self.value.bias""",
        )     )
    rename_keys.append(
        (
            F"""{old}/encoder/layer_._{i}/attention/output/dense/kernel:0""",
            F"""{new}.encoder.layer.{i}.attention.output.dense.weight""",
        )     )
    rename_keys.append(
        (
            F"""{old}/encoder/layer_._{i}/attention/output/dense/bias:0""",
            F"""{new}.encoder.layer.{i}.attention.output.dense.bias""",
        )     )
    rename_keys.append(
        (
            F"""{old}/encoder/layer_._{i}/attention/output/LayerNorm/gamma:0""",
            F"""{new}.encoder.layer.{i}.attention.output.LayerNorm.weight""",
        )     )
    rename_keys.append(
        (
            F"""{old}/encoder/layer_._{i}/attention/output/LayerNorm/beta:0""",
            F"""{new}.encoder.layer.{i}.attention.output.LayerNorm.bias""",
        )     )
    rename_keys.append(
        (
            F"""{old}/encoder/layer_._{i}/intermediate/dense/kernel:0""",
            F"""{new}.encoder.layer.{i}.intermediate.dense.weight""",
        )     )
    rename_keys.append(
        (
            F"""{old}/encoder/layer_._{i}/intermediate/dense/bias:0""",
            F"""{new}.encoder.layer.{i}.intermediate.dense.bias""",
        )     )
    rename_keys.append(
        (F"""{old}/encoder/layer_._{i}/output/dense/kernel:0""", F"""{new}.encoder.layer.{i}.output.dense.weight""")     )
    rename_keys.append(
        (F"""{old}/encoder/layer_._{i}/output/dense/bias:0""", F"""{new}.encoder.layer.{i}.output.dense.bias""")     )
    rename_keys.append(
        (F"""{old}/encoder/layer_._{i}/output/LayerNorm/gamma:0""", F"""{new}.encoder.layer.{i}.output.LayerNorm.weight""")     )
    rename_keys.append(
        (F"""{old}/encoder/layer_._{i}/output/LayerNorm/beta:0""", F"""{new}.encoder.layer.{i}.output.LayerNorm.bias""")     )

  rename_keys.append((F"""{old}/embeddings/word_embeddings/weight:0""", F"""{new}.embeddings.word_embeddings.weight""")     )
  rename_keys.append(
      (F"""{old}/embeddings/position_embeddings/embeddings:0""", F"""{new}.embeddings.position_embeddings.weight""")     )
  rename_keys.append(
      (F"""{old}/embeddings/token_type_embeddings/embeddings:0""", F"""{new}.embeddings.token_type_embeddings.weight""")     )
  rename_keys.append((F"""{old}/embeddings/LayerNorm/gamma:0""", F"""{new}.embeddings.LayerNorm.weight""")     )
  rename_keys.append((F"""{old}/embeddings/LayerNorm/beta:0""", F"""{new}.embeddings.LayerNorm.bias""")     )

  rename_keys.append((F"""{old}/pooler/dense/kernel:0""", F"""{new}.pooler.dense.weight""")     )
  rename_keys.append((F"""{old}/pooler/dense/bias:0""", F"""{new}.pooler.dense.bias""")     )
  rename_keys.append(('''dense/kernel:0''', '''text_projection.weight''')     )
  rename_keys.append(('''dense/bias:0''', '''text_projection.bias''')     )
  rename_keys.append(('''dense/bias:0''', '''text_projection.bias''')     )
  rename_keys.append(('''temperature:0''', '''temperature''')     )

  for item in rename_keys:
    if item[0] in original_param_names:
      _snake_case						=							item[1]
  return key_mapping



def       snake_case_(_UpperCamelCase ,       _UpperCamelCase ,       _UpperCamelCase     )							->	Any:



  """simple docstring"""






  list(hf_params.keys()     )

  for key, value in tf_params.items():
    if key not in key_mapping:
      continue

    _snake_case						=							key_mapping[key]
    if "_conv" in key and "kernel" in key:
      _snake_case						=							torch.from_numpy(_SCREAMING_SNAKE_CASE     ).permute(3 ,       2 ,       0 ,       1     )
    elif "embeddings" in key:
      _snake_case						=							torch.from_numpy(_SCREAMING_SNAKE_CASE     )
    elif "depthwise_kernel" in key:
      _snake_case						=							torch.from_numpy(_SCREAMING_SNAKE_CASE     ).permute(2 ,       3 ,       0 ,       1     )
    elif "kernel" in key:
      _snake_case						=							torch.from_numpy(np.transpose(_SCREAMING_SNAKE_CASE     )     )
    elif "temperature" in key:
      _snake_case						=							value
    elif "bn/gamma" or "bn/beta" in key:
      _snake_case						=							torch.from_numpy(np.transpose(_SCREAMING_SNAKE_CASE     )     ).squeeze()
    else:
      _snake_case						=							torch.from_numpy(_SCREAMING_SNAKE_CASE     )

    # Replace HF parameters with original TF model parameters
    hf_params[hf_key].copy_(_SCREAMING_SNAKE_CASE     )



@torch.no_grad()
def       snake_case_(_UpperCamelCase ,       _UpperCamelCase ,       _UpperCamelCase ,       _UpperCamelCase     )							->	List[str]:



  """simple docstring"""






  _snake_case						=							64
  _snake_case						=							Tokenizer(_SCREAMING_SNAKE_CASE     )
  _snake_case						=							align.Align('''efficientnet-b7''' ,       '''bert-base''' ,       640 ,       _SCREAMING_SNAKE_CASE ,       tok.get_vocab_size()     )
  original_model.compile()
  original_model.load_weights(_SCREAMING_SNAKE_CASE     )

  _snake_case						=							original_model.trainable_variables
  _snake_case						=							original_model.non_trainable_variables
  _snake_case						=							{param.name: param.numpy() for param in tf_params}
  for param in tf_non_train_params:
    _snake_case						=							param.numpy()
  _snake_case						=							list(tf_params.keys()     )

  # Load HuggingFace model
  _snake_case						=							get_align_config()
  _snake_case						=							AlignModel(_SCREAMING_SNAKE_CASE     ).eval()
  _snake_case						=							hf_model.state_dict()

  # Create src-to-dst parameter name mapping dictionary
  print('''Converting parameters...'''     )
  _snake_case						=							rename_keys(_SCREAMING_SNAKE_CASE     )
  replace_params(_SCREAMING_SNAKE_CASE ,       _SCREAMING_SNAKE_CASE ,       _SCREAMING_SNAKE_CASE     )

  # Initialize processor
  _snake_case						=							get_processor()
  _snake_case						=							processor(
      images=prepare_img() ,       text='''A picture of a cat''' ,       padding='''max_length''' ,       max_length=64 ,       return_tensors='''pt'''     )

  # HF model inference
  hf_model.eval()
  with torch.no_grad():
    _snake_case						=							hf_model(**_SCREAMING_SNAKE_CASE     )

  _snake_case						=							outputs.image_embeds.detach().numpy()
  _snake_case						=							outputs.text_embeds.detach().numpy()

  # Original model inference
  _snake_case						=							False
  _snake_case						=							EfficientNetImageProcessor(
      do_center_crop=_SCREAMING_SNAKE_CASE ,       do_rescale=_SCREAMING_SNAKE_CASE ,       do_normalize=_SCREAMING_SNAKE_CASE ,       include_top=_SCREAMING_SNAKE_CASE ,       resample=Image.BILINEAR ,       )
  _snake_case						=							tf_image_processor(images=prepare_img() ,       return_tensors='''tf''' ,       data_format='''channels_last'''     )["pixel_values"]
  _snake_case						=							tok(tf.constant(['''A picture of a cat''']     )     )

  _snake_case						=							original_model.image_encoder(_SCREAMING_SNAKE_CASE ,       training=_SCREAMING_SNAKE_CASE     )
  _snake_case						=							original_model.text_encoder(_SCREAMING_SNAKE_CASE ,       training=_SCREAMING_SNAKE_CASE     )

  _snake_case						=							tf.nn.la_normalize(_SCREAMING_SNAKE_CASE ,       axis=-1     )
  _snake_case						=							tf.nn.la_normalize(_SCREAMING_SNAKE_CASE ,       axis=-1     )

  # Check whether original and HF model outputs match  -> np.allclose
  if not np.allclose(_SCREAMING_SNAKE_CASE ,       _SCREAMING_SNAKE_CASE ,       atol=1E-3     ):
    raise ValueError('''The predicted image features are not the same.'''     )
  if not np.allclose(_SCREAMING_SNAKE_CASE ,       _SCREAMING_SNAKE_CASE ,       atol=1E-3     ):
    raise ValueError('''The predicted text features are not the same.'''     )
  print('''Model outputs match!'''     )

  if save_model:
    # Create folder to save model
    if not os.path.isdir(_SCREAMING_SNAKE_CASE     ):
      os.mkdir(_SCREAMING_SNAKE_CASE     )
    # Save converted model and image processor
    hf_model.save_pretrained(_SCREAMING_SNAKE_CASE     )
    processor.save_pretrained(_SCREAMING_SNAKE_CASE     )

  if push_to_hub:
    # Push model and image processor to hub
    print('''Pushing converted ALIGN to the hub...'''     )
    processor.push_to_hub('''align-base'''     )
    hf_model.push_to_hub('''align-base'''     )


if __name__ == "__main__":
 __A					=	argparse.ArgumentParser()
 # Required parameters
 parser.add_argument(
     '''--checkpoint_path''',
     default='''./weights/model-weights''',
     type=str,
     help='''Path to the pretrained TF ALIGN checkpoint.''',
 )
 parser.add_argument(
     '''--pytorch_dump_folder_path''',
     default='''hf_model''',
     type=str,
     help='''Path to the output PyTorch model directory.''',
 )
 parser.add_argument('''--save_model''', action='''store_true''', help='''Save model to local''')
 parser.add_argument('''--push_to_hub''', action='''store_true''', help='''Push model and image processor to the hub''')

 __A					=	parser.parse_args()
 convert_align_checkpoint(args.checkpoint_path, args.pytorch_dump_folder_path, args.save_model, args.push_to_hub)




'''simple docstring'''
from transformers import PretrainedConfig
class  _snake_case						(__SCREAMING_SNAKE_CASE):
			__A   :     Optional[Any]          ="custom"






			def __init__(   self       ,_snake_case=1       ,**_snake_case			):
				UpperCAmelCase_			:	Optional[int]      =   attribute
				super().__init__(**_snake_case			)




class  _snake_case						(__SCREAMING_SNAKE_CASE):
			__A   :     Dict          ="custom"






			def __init__(   self       ,_snake_case=1       ,**_snake_case			):
				UpperCAmelCase_			:	Optional[int]      =   attribute




