


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()
__lowerCAmelCase									=				logging.get_logger(__name__)




def   snake_case_	(    snake_case	)    ->							str:
       lowercase__:  Dict         =  tf.image.resize(snake_case     ,				(3_46, 3_46)	)
       lowercase__:  str         =  tf.image.crop_to_bounding_box(snake_case     ,				(3_46 - 2_89) // 2     ,				(3_46 - 2_89) // 2     ,				2_89     ,				2_89	)
       return image




def   snake_case_	(    )    ->							str:
       lowercase__:  List[str]         =  EfficientNetConfig.from_pretrained('google/efficientnet-b7'	)
       lowercase__:  List[str]         =  2_89
       lowercase__:  Any         =  6_40
       lowercase__:  str         =  {'0': 'LABEL_0', '1': 'LABEL_1'}
       lowercase__:  str         =  {'LABEL_0': 0, 'LABEL_1': 1}
       lowercase__:  List[str]         =  []

       lowercase__:  List[str]         =  BertConfig()
       lowercase__:  List[str]         =  AlignConfig.from_text_vision_configs(
           text_config=snake_case     ,				vision_config=snake_case     ,				projection_dim=6_40	)
       return config




def   snake_case_	(    )    ->							List[str]:
       lowercase__:  Any         =  'http://images.cocodataset.org/val2017/000000039769.jpg'
       lowercase__:  Union[str, Any]         =  Image.open(requests.get(snake_case     ,				stream=snake_case	).raw	)
       return im




def   snake_case_	(    )    ->							Optional[int]:
       lowercase__:  Dict         =  EfficientNetImageProcessor(
           do_center_crop=snake_case     ,				rescale_factor=1 / 1_2_7.5     ,				rescale_offset=snake_case     ,				do_normalize=snake_case     ,				include_top=snake_case     ,				resample=Image.BILINEAR     ,				)
       lowercase__:  Tuple         =  BertTokenizer.from_pretrained('bert-base-uncased'	)
       lowercase__:  Tuple         =  64
       lowercase__:  str         =  AlignProcessor(image_processor=snake_case     ,				tokenizer=snake_case	)
       return processor




def   snake_case_	(    snake_case	)    ->							Tuple:
       # EfficientNet image encoder
       lowercase__:  int         =  [v.split('_'	)[0].split('block'	)[1] for v in original_param_names if v.startswith('block'	)]
       lowercase__:  List[Any]         =  list(set(snake_case	)	)
       lowercase__:  Dict         =  sorted(snake_case	)
       lowercase__:  Union[str, Any]         =  len(snake_case	)
       lowercase__:  int         =  {b: str(snake_case	) for b, i in zip(snake_case     ,				range(snake_case	)	)}

       lowercase__:  Dict         =  []
       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:
              lowercase__:  Dict         =  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')	)

       lowercase__:  Any         =  {}
       for item in rename_keys:
              if item[0] in original_param_names:
                     lowercase__:  Union[str, Any]         =  'vision_model.' + item[1]

    # BERT text encoder
       lowercase__:  Optional[int]         =  []
       lowercase__:  List[str]         =  'tf_bert_model/bert'
       lowercase__:  Optional[Any]         =  '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:
                     lowercase__:  Optional[int]         =  item[1]
       return key_mapping




def   snake_case_	(    snake_case     ,				snake_case     ,				snake_case	)    ->							List[Any]:
       list(hf_params.keys()	)

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

              lowercase__:  Optional[Any]         =  key_mapping[key]
              if "_conv" in key and "kernel" in key:
                     lowercase__:  Dict         =  torch.from_numpy(snake_case	).permute(3     ,				2     ,				0     ,				1	)
              elif "embeddings" in key:
                     lowercase__:  Optional[Any]         =  torch.from_numpy(snake_case	)
              elif "depthwise_kernel" in key:
                     lowercase__:  Optional[int]         =  torch.from_numpy(snake_case	).permute(2     ,				3     ,				0     ,				1	)
              elif "kernel" in key:
                     lowercase__:  Optional[Any]         =  torch.from_numpy(np.transpose(snake_case	)	)
              elif "temperature" in key:
                     lowercase__:  Union[str, Any]         =  value
              elif "bn/gamma" or "bn/beta" in key:
                     lowercase__:  Tuple         =  torch.from_numpy(np.transpose(snake_case	)	).squeeze()
              else:
                     lowercase__:  List[str]         =  torch.from_numpy(snake_case	)

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




@torch.no_grad()
def   snake_case_	(    snake_case     ,				snake_case     ,				snake_case     ,				snake_case	)    ->							Optional[int]:
       lowercase__:  List[Any]         =  64
       lowercase__:  Dict         =  Tokenizer(snake_case	)
       lowercase__:  Optional[int]         =  align.Align('efficientnet-b7'     ,				'bert-base'     ,				6_40     ,				snake_case     ,				tok.get_vocab_size()	)
       original_model.compile()
       original_model.load_weights(snake_case	)

       lowercase__:  str         =  original_model.trainable_variables
       lowercase__:  List[Any]         =  original_model.non_trainable_variables
       lowercase__:  Union[str, Any]         =  {param.name: param.numpy() for param in tf_params}
       for param in tf_non_train_params:
              lowercase__:  Dict         =  param.numpy()
       lowercase__:  int         =  list(tf_params.keys()	)

       # Load HuggingFace model
       lowercase__:  str         =  get_align_config()
       lowercase__:  Optional[int]         =  AlignModel(snake_case	).eval()
       lowercase__:  Dict         =  hf_model.state_dict()

       # Create src-to-dst parameter name mapping dictionary
       print('Converting parameters...'	)
       lowercase__:  List[str]         =  rename_keys(snake_case	)
       replace_params(snake_case     ,				snake_case     ,				snake_case	)

       # Initialize processor
       lowercase__:  List[Any]         =  get_processor()
       lowercase__:  Dict         =  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():
              lowercase__:  Tuple         =  hf_model(**snake_case	)

       lowercase__:  Dict         =  outputs.image_embeds.detach().numpy()
       lowercase__:  List[Any]         =  outputs.text_embeds.detach().numpy()

       # Original model inference
       lowercase__:  List[Any]         =  False
       lowercase__:  int         =  EfficientNetImageProcessor(
           do_center_crop=snake_case     ,				do_rescale=snake_case     ,				do_normalize=snake_case     ,				include_top=snake_case     ,				resample=Image.BILINEAR     ,				)
       lowercase__:  Any         =  tf_image_processor(images=prepare_img()     ,				return_tensors='tf'     ,				data_format='channels_last'	)['pixel_values']
       lowercase__:  Union[str, Any]         =  tok(tf.constant(['A picture of a cat']	)	)

       lowercase__:  int         =  original_model.image_encoder(snake_case     ,				training=snake_case	)
       lowercase__:  List[str]         =  original_model.text_encoder(snake_case     ,				training=snake_case	)

       lowercase__:  Optional[int]         =  tf.nn.la_normalize(snake_case     ,				axis=-1	)
       lowercase__:  List[Any]         =  tf.nn.la_normalize(snake_case     ,				axis=-1	)

       # Check whether original and HF model outputs match  -> np.allclose
       if not np.allclose(snake_case     ,				snake_case     ,				atol=1e-3	):
              raise ValueError('The predicted image features are not the same.'	)
       if not np.allclose(snake_case     ,				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(snake_case	):
                     os.mkdir(snake_case	)
              # Save converted model and image processor
              hf_model.save_pretrained(snake_case	)
              processor.save_pretrained(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__":
 __lowerCAmelCase									=				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''')

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






class     __a     :
 def __init__(  self					)   ->							Tuple:




        '''simple docstring'''
        lowercase__:  str         =  []
        lowercase__:  Any         =  0
        lowercase__:  int         =  0
 def __str__(  self					)   ->							Any:




        '''simple docstring'''
        lowercase__:  str         =  '<' + str(self.entries					)[1:-1] + '>'
        return printed
 def 						SCREAMING_SNAKE_CASE__  (  self							,					lowerCAmelCase__					)   ->							List[str]:




        '''simple docstring'''
        self.entries.append(lowerCAmelCase__					)
        lowercase__:  List[str]         =  self.length + 1
 def 						SCREAMING_SNAKE_CASE__  (  self					)   ->							str:




        '''simple docstring'''
        lowercase__:  Any         =  self.length - 1
        lowercase__:  Optional[int]         =  self.entries[self.front]
        # self.front-=1
        # self.entries = self.entries[self.front:]
        lowercase__:  int         =  self.entries[1:]
        return dequeued
 def 						SCREAMING_SNAKE_CASE__  (  self							,					lowerCAmelCase__					)   ->							List[str]:




        '''simple docstring'''
        for _ in range(lowerCAmelCase__					):
               self.put(self.get()					)
 def 						SCREAMING_SNAKE_CASE__  (  self					)   ->							Union[str, Any]:




        '''simple docstring'''
        return self.entries[0]
 def 						SCREAMING_SNAKE_CASE__  (  self					)   ->							Optional[int]:




        '''simple docstring'''
        return self.length


