

from typing import TYPE_CHECKING

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


lowerCAmelCase       :				Union[str, Any]         =					{'configuration_cvt': ['CVT_PRETRAINED_CONFIG_ARCHIVE_MAP', 'CvtConfig']}


try:
    if not is_torch_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowerCAmelCase       :				List[Any]         =					[
        'CVT_PRETRAINED_MODEL_ARCHIVE_LIST',
        'CvtForImageClassification',
        'CvtModel',
        'CvtPreTrainedModel',
    ]

try:
    if not is_tf_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    lowerCAmelCase       :				Union[str, Any]         =					[
        'TF_CVT_PRETRAINED_MODEL_ARCHIVE_LIST',
        'TFCvtForImageClassification',
        'TFCvtModel',
        'TFCvtPreTrainedModel',
    ]

if TYPE_CHECKING:
    from .configuration_cvt import CVT_PRETRAINED_CONFIG_ARCHIVE_MAP, CvtConfig

    try:
        if not is_torch_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_cvt import (
            CVT_PRETRAINED_MODEL_ARCHIVE_LIST,
            CvtForImageClassification,
            CvtModel,
            CvtPreTrainedModel,
        )

    try:
        if not is_tf_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_tf_cvt import (
            TF_CVT_PRETRAINED_MODEL_ARCHIVE_LIST,
            TFCvtForImageClassification,
            TFCvtModel,
            TFCvtPreTrainedModel,
        )


else:
    import sys

    lowerCAmelCase       :				Any         =					_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)









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

def 				A_					(							a					):

      """simple docstring"""



      SCREAMING_SNAKE_CASE_			:					str          =       tf.image.resize(a   ,				(3_4_6, 3_4_6)					)
      SCREAMING_SNAKE_CASE_			:					Any          =       tf.image.crop_to_bounding_box(a   ,				(3_4_6 - 2_8_9) // 2   ,				(3_4_6 - 2_8_9) // 2   ,				2_8_9   ,				2_8_9					)
      return image

def 				A_					(							):

      """simple docstring"""



      SCREAMING_SNAKE_CASE_			:					List[str]          =       EfficientNetConfig.from_pretrained('google/efficientnet-b7'					)
      SCREAMING_SNAKE_CASE_			:					int          =       2_8_9
      SCREAMING_SNAKE_CASE_			:					int          =       6_4_0
      SCREAMING_SNAKE_CASE_			:					List[Any]          =       {'0': 'LABEL_0', '1': 'LABEL_1'}
      SCREAMING_SNAKE_CASE_			:					Tuple          =       {'LABEL_0': 0, 'LABEL_1': 1}
      SCREAMING_SNAKE_CASE_			:					Optional[int]          =       []

      SCREAMING_SNAKE_CASE_			:					Optional[Any]          =       BertConfig()
      SCREAMING_SNAKE_CASE_			:					Dict          =       AlignConfig.from_text_vision_configs(
          text_config=a   ,				vision_config=a   ,				projection_dim=6_4_0					)
      return config

def 				A_					(							):

      """simple docstring"""



      SCREAMING_SNAKE_CASE_			:					Dict          =       'http://images.cocodataset.org/val2017/000000039769.jpg'
      SCREAMING_SNAKE_CASE_			:					Optional[Any]          =       Image.open(requests.get(a   ,				stream=a					).raw					)
      return im

def 				A_					(							):

      """simple docstring"""



      SCREAMING_SNAKE_CASE_			:					int          =       EfficientNetImageProcessor(
          do_center_crop=a   ,				rescale_factor=1 / 1_27.5   ,				rescale_offset=a   ,				do_normalize=a   ,				include_top=a   ,				resample=Image.BILINEAR   ,				)
      SCREAMING_SNAKE_CASE_			:					List[str]          =       BertTokenizer.from_pretrained('bert-base-uncased'					)
      SCREAMING_SNAKE_CASE_			:					Tuple          =       6_4
      SCREAMING_SNAKE_CASE_			:					Tuple          =       AlignProcessor(image_processor=a   ,				tokenizer=a					)
      return processor

def 				A_					(							a					):

      """simple docstring"""



      SCREAMING_SNAKE_CASE_			:					Any          =       [v.split('_'					)[0].split('block'					)[1] for v in original_param_names if v.startswith('block'					)]
      SCREAMING_SNAKE_CASE_			:					Any          =       list(set(a					)					)
      SCREAMING_SNAKE_CASE_			:					List[Any]          =       sorted(a					)
      SCREAMING_SNAKE_CASE_			:					Union[str, Any]          =       len(a					)
      SCREAMING_SNAKE_CASE_			:					int          =       {b: str(a					) for b, i in zip(a   ,				range(a					)					)}

      SCREAMING_SNAKE_CASE_			:					str          =       []
      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:
            SCREAMING_SNAKE_CASE_			:					Optional[int]          =       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")					)

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

    # BERT text encoder
      SCREAMING_SNAKE_CASE_			:					Optional[Any]          =       []
      SCREAMING_SNAKE_CASE_			:					Tuple          =       'tf_bert_model/bert'
      SCREAMING_SNAKE_CASE_			:					Any          =       'text_model'
      for i in range(1_2					):
            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:
                  SCREAMING_SNAKE_CASE_			:					int          =       item[1]
      return key_mapping

def 				A_					(							a   ,				a   ,				a					):

      """simple docstring"""



      list(hf_params.keys()					)

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

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

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

@torch.no_grad()
def 				A_					(							a   ,				a   ,				a   ,				a					):

      """simple docstring"""



      SCREAMING_SNAKE_CASE_			:					Optional[int]          =       6_4
      SCREAMING_SNAKE_CASE_			:					List[str]          =       Tokenizer(a					)
      SCREAMING_SNAKE_CASE_			:					Optional[Any]          =       align.Align('efficientnet-b7'   ,				'bert-base'   ,				6_4_0   ,				a   ,				tok.get_vocab_size()					)
      original_model.compile()
      original_model.load_weights(a					)

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

      # Load HuggingFace model
      SCREAMING_SNAKE_CASE_			:					Any          =       get_align_config()
      SCREAMING_SNAKE_CASE_			:					Union[str, Any]          =       AlignModel(a					).eval()
      SCREAMING_SNAKE_CASE_			:					Optional[int]          =       hf_model.state_dict()

      # Create src-to-dst parameter name mapping dictionary
      print('Converting parameters...'					)
      SCREAMING_SNAKE_CASE_			:					str          =       rename_keys(a					)
      replace_params(a   ,				a   ,				a					)

      # Initialize processor
      SCREAMING_SNAKE_CASE_			:					Optional[Any]          =       get_processor()
      SCREAMING_SNAKE_CASE_			:					List[str]          =       processor(
          images=prepare_img()   ,				text='A picture of a cat'   ,				padding='max_length'   ,				max_length=6_4   ,				return_tensors='pt'					)

      # HF model inference
      hf_model.eval()
      with torch.no_grad():
            SCREAMING_SNAKE_CASE_			:					List[Any]          =       hf_model(**a					)

      SCREAMING_SNAKE_CASE_			:					Optional[int]          =       outputs.image_embeds.detach().numpy()
      SCREAMING_SNAKE_CASE_			:					Optional[Any]          =       outputs.text_embeds.detach().numpy()

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

      SCREAMING_SNAKE_CASE_			:					List[str]          =       original_model.image_encoder(a   ,				training=a					)
      SCREAMING_SNAKE_CASE_			:					Any          =       original_model.text_encoder(a   ,				training=a					)

      SCREAMING_SNAKE_CASE_			:					str          =       tf.nn.la_normalize(a   ,				axis=-1					)
      SCREAMING_SNAKE_CASE_			:					int          =       tf.nn.la_normalize(a   ,				axis=-1					)

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

      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       :				str         =					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       :				str         =					parser.parse_args()
    convert_align_checkpoint(args.checkpoint_path, args.pytorch_dump_folder_path, args.save_model, args.push_to_hub)






