


import argparse
import json
import os

import numpy as np
import PIL
import requests
import tensorflow.keras.applications.efficientnet as efficientnet
import torch
from huggingface_hub import hf_hub_download
from PIL import Image
from tensorflow.keras.preprocessing import image

from transformers import (
    EfficientNetConfig,
    EfficientNetForImageClassification,
    EfficientNetImageProcessor,
)
from transformers.utils import logging


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

_A   =			{
    "b0": efficientnet.EfficientNetBa,
    "b1": efficientnet.EfficientNetBa,
    "b2": efficientnet.EfficientNetBa,
    "b3": efficientnet.EfficientNetBa,
    "b4": efficientnet.EfficientNetBa,
    "b5": efficientnet.EfficientNetBa,
    "b6": efficientnet.EfficientNetBa,
    "b7": efficientnet.EfficientNetBa,
}

_A   =			{
    "b0": {
        "hidden_dim": 12_80,
        "width_coef": 1.0,
        "depth_coef": 1.0,
        "image_size": 2_24,
        "dropout_rate": 0.2,
        "dw_padding": [],
    },
    "b1": {
        "hidden_dim": 12_80,
        "width_coef": 1.0,
        "depth_coef": 1.1,
        "image_size": 2_40,
        "dropout_rate": 0.2,
        "dw_padding": [16],
    },
    "b2": {
        "hidden_dim": 14_08,
        "width_coef": 1.1,
        "depth_coef": 1.2,
        "image_size": 2_60,
        "dropout_rate": 0.3,
        "dw_padding": [5, 8, 16],
    },
    "b3": {
        "hidden_dim": 15_36,
        "width_coef": 1.2,
        "depth_coef": 1.4,
        "image_size": 3_00,
        "dropout_rate": 0.3,
        "dw_padding": [5, 18],
    },
    "b4": {
        "hidden_dim": 17_92,
        "width_coef": 1.4,
        "depth_coef": 1.8,
        "image_size": 3_80,
        "dropout_rate": 0.4,
        "dw_padding": [6],
    },
    "b5": {
        "hidden_dim": 20_48,
        "width_coef": 1.6,
        "depth_coef": 2.2,
        "image_size": 4_56,
        "dropout_rate": 0.4,
        "dw_padding": [13, 27],
    },
    "b6": {
        "hidden_dim": 23_04,
        "width_coef": 1.8,
        "depth_coef": 2.6,
        "image_size": 5_28,
        "dropout_rate": 0.5,
        "dw_padding": [31],
    },
    "b7": {
        "hidden_dim": 25_60,
        "width_coef": 2.0,
        "depth_coef": 3.1,
        "image_size": 6_00,
        "dropout_rate": 0.5,
        "dw_padding": [18],
    },
}



def 						lowercase_      (       A__		)							->							int:
    """simple docstring"""

    snake_case								= EfficientNetConfig()
    snake_case								= CONFIG_MAP[model_name]["hidden_dim"]
    snake_case								= CONFIG_MAP[model_name]["width_coef"]
    snake_case								= CONFIG_MAP[model_name]["depth_coef"]
    snake_case								= CONFIG_MAP[model_name]["image_size"]
    snake_case								= CONFIG_MAP[model_name]["dropout_rate"]
    snake_case								= CONFIG_MAP[model_name]["dw_padding"]

    snake_case								= "huggingface/label-files"
    snake_case								= "imagenet-1k-id2label.json"
    snake_case								= 1000
    snake_case								= json.load(open(hf_hub_download(A__			,		A__			,		repo_type="dataset"		)			,		"r"		)		)
    snake_case								= {int(A__		): v for k, v in idalabel.items()}

    snake_case								= idalabel
    snake_case								= {v: k for k, v in idalabel.items()}
    return config



def 						lowercase_      (       )							->							str:
    """simple docstring"""

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



def 						lowercase_      (       A__		)							->							Tuple:
    """simple docstring"""

    snake_case								= CONFIG_MAP[model_name]["image_size"]
    snake_case								= EfficientNetImageProcessor(
        size={"height": size, "width": size}			,		image_mean=[0.485, 0.456, 0.406]			,		image_std=[0.47853944, 0.4732864, 0.47434163]			,		do_center_crop=A__			,		)
    return preprocessor



def 						lowercase_      (       A__		)							->							Optional[int]:
    """simple docstring"""

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

    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')		)

    rename_keys.append(("top_conv/kernel:0", "encoder.top_conv.weight")		)
    rename_keys.append(("top_bn/gamma:0", "encoder.top_bn.weight")		)
    rename_keys.append(("top_bn/beta:0", "encoder.top_bn.bias")		)
    rename_keys.append(("top_bn/moving_mean:0", "encoder.top_bn.running_mean")		)
    rename_keys.append(("top_bn/moving_variance:0", "encoder.top_bn.running_var")		)

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

    snake_case								= "classifier.weight"
    snake_case								= "classifier.bias"
    return key_mapping



def 						lowercase_      (       A__			,		A__			,		A__		)							->							Optional[Any]:
    """simple docstring"""

    for key, value in tf_params.items():
        if "normalization" in key:
            continue

        snake_case								= key_mapping[key]
        if "_conv" in key and "kernel" in key:
            snake_case								= torch.from_numpy(A__		).permute(3			,		2			,		0			,		1		)
        elif "depthwise_kernel" in key:
            snake_case								= torch.from_numpy(A__		).permute(2			,		3			,		0			,		1		)
        elif "kernel" in key:
            snake_case								= torch.from_numpy(np.transpose(A__		)		)
        else:
            snake_case								= torch.from_numpy(A__		)

        # Replace HF parameters with original TF model parameters
        assert hf_params[hf_key].shape == new_hf_value.shape
        hf_params[hf_key].copy_(A__		)



@torch.no_grad()
def 						lowercase_      (       A__			,		A__			,		A__			,		A__		)							->							Union[str, Any]:
    """simple docstring"""

    snake_case								= model_classes[model_name](
        include_top=A__			,		weights="imagenet"			,		input_tensor=A__			,		input_shape=A__			,		pooling=A__			,		classes=1000			,		classifier_activation="softmax"			,		)

    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_efficientnet_config(A__		)
    snake_case								= EfficientNetForImageClassification(A__		).eval()
    snake_case								= hf_model.state_dict()

    # Create src-to-dst parameter name mapping dictionary
    print("Converting parameters..."		)
    snake_case								= rename_keys(A__		)
    replace_params(A__			,		A__			,		A__		)

    # Initialize preprocessor and preprocess input image
    snake_case								= convert_image_processor(A__		)
    snake_case								= preprocessor(images=prepare_img()			,		return_tensors="pt"		)

    # HF model inference
    hf_model.eval()
    with torch.no_grad():
        snake_case								= hf_model(**A__		)
    snake_case								= outputs.logits.detach().numpy()

    # Original model inference
    snake_case								= False
    snake_case								= CONFIG_MAP[model_name]["image_size"]
    snake_case								= prepare_img().resize((image_size, image_size)			,		resample=PIL.Image.NEAREST		)
    snake_case								= image.img_to_array(A__		)
    snake_case								= np.expand_dims(A__			,		axis=0		)
    snake_case								= original_model.predict(A__		)

    # Check whether original and HF model outputs match  -> np.allclose
    assert np.allclose(A__			,		A__			,		atol=1e-3		), "The predicted logits 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__		)
        preprocessor.save_pretrained(A__		)

    if push_to_hub:
        # Push model and image processor to hub
        print(F'Pushing converted {model_name} to the hub...'		)
        snake_case								= F'efficientnet-{model_name}'
        preprocessor.push_to_hub(A__		)
        hf_model.push_to_hub(A__		)


if __name__ == "__main__":
   _A   =			argparse.ArgumentParser()
   # Required parameters
   parser.add_argument(
       "--model_name",
       default="b0",
       type=str,
       help="Version name of the EfficientNet model you want to convert, select from [b0, b1, b2, b3, b4, b5, b6, b7].",
   )
   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_efficientnet_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.save_model, args.push_to_hub)



import importlib.metadata
import warnings
from copy import deepcopy

from packaging import version

from ..utils import logging
from .import_utils import is_accelerate_available, is_bitsandbytes_available


if is_bitsandbytes_available():
   import bitsandbytes as bnb
   import torch
   import torch.nn as nn

   from ..pytorch_utils import ConvaD

if is_accelerate_available():
   from accelerate import init_empty_weights
   from accelerate.utils import find_tied_parameters

_A   =			logging.get_logger(__name__)



def 						lowercase_      (       A__			,		A__			,		A__			,		A__=None			,		A__=None		)							->							str:
    """simple docstring"""

    if "." in tensor_name:
        snake_case								= tensor_name.split("."		)
        for split in splits[:-1]:
            snake_case								= getattr(A__			,		A__		)
            if new_module is None:
                raise ValueError(F'{module} has no attribute {split}.'		)
            snake_case								= new_module
        snake_case								= splits[-1]

    if tensor_name not in module._parameters and tensor_name not in module._buffers:
        raise ValueError(F'{module} does not have a parameter or a buffer named {tensor_name}.'		)
    snake_case								= tensor_name in module._buffers
    snake_case								= getattr(A__			,		A__		)

    if old_value.device == torch.device("meta"		) and device not in ["meta", torch.device("meta"		)] and value is None:
        raise ValueError(F'{tensor_name} is on the meta device, we need a `value` to put in on {device}.'		)

    snake_case								= False
    snake_case								= False
    if is_buffer or not is_bitsandbytes_available():
        snake_case								= False
        snake_case								= False
    else:
        snake_case								= hasattr(bnb.nn			,		"Params4bit"		) and isinstance(module._parameters[tensor_name]			,		bnb.nn.Paramsabit		)
        snake_case								= isinstance(module._parameters[tensor_name]			,		bnb.nn.IntaParams		)

    if is_abit or is_abit:
        snake_case								= module._parameters[tensor_name]
        if param.device.type != "cuda":
            if value is None:
                snake_case								= old_value.to(A__		)
            elif isinstance(A__			,		torch.Tensor		):
                snake_case								= value.to("cpu"		)
                if value.dtype == torch.inta:
                    snake_case								= version.parse(importlib.metadata.version("bitsandbytes"		)		) > version.parse(
                        "0.37.2"		)
                    if not is_abit_serializable:
                        raise ValueError(
                            "Detected int8 weights but the version of bitsandbytes is not compatible with int8 serialization. "
                            "Make sure to download the latest `bitsandbytes` version. `pip install --upgrade bitsandbytes`."		)
            else:
                snake_case								= torch.tensor(A__			,		device="cpu"		)

            # Support models using `Conv1D` in place of `nn.Linear` (e.g. gpt2) by transposing the weight matrix prior to quantization.
            # Since weights are saved in the correct "orientation", we skip transposing when loading.
            if issubclass(module.source_cls			,		A__		) and fpaa_statistics is None:
                snake_case								= new_value.T

            snake_case								= old_value.__dict__
            if is_abit:
                snake_case								= bnb.nn.IntaParams(A__			,		requires_grad=A__			,		**A__		).to(A__		)
            elif is_abit:
                snake_case								= bnb.nn.Paramsabit(A__			,		requires_grad=A__			,		**A__		).to(A__		)

            snake_case								= new_value
            if fpaa_statistics is not None:
                setattr(module.weight			,		"SCB"			,		fpaa_statistics.to(A__		)		)

    else:
        if value is None:
            snake_case								= old_value.to(A__		)
        elif isinstance(A__			,		torch.Tensor		):
            snake_case								= value.to(A__		)
        else:
            snake_case								= torch.tensor(A__			,		device=A__		)

        if is_buffer:
            snake_case								= new_value
        else:
            snake_case								= nn.Parameter(A__			,		requires_grad=old_value.requires_grad		)
            snake_case								= new_value



def 						lowercase_      (       A__			,		A__=None			,		A__=None			,		A__=None			,		A__=False		)							->							Optional[Any]:
    """simple docstring"""

    for name, module in model.named_children():
        if current_key_name is None:
            snake_case								= []
        current_key_name.append(A__		)

        if (isinstance(A__			,		nn.Linear		) or isinstance(A__			,		A__		)) and name not in modules_to_not_convert:
            # Check if the current key is not in the `modules_to_not_convert`
            if not any(key in ".".join(A__		) for key in modules_to_not_convert		):
                with init_empty_weights():
                    if isinstance(A__			,		A__		):
                        snake_case					,							snake_case								= module.weight.shape
                    else:
                        snake_case								= module.in_features
                        snake_case								= module.out_features

                    if quantization_config.quantization_method() == "llm_int8":
                        snake_case								= bnb.nn.LinearabitLt(
                            A__			,		A__			,		module.bias is not None			,		has_fpaa_weights=quantization_config.llm_inta_has_fpaa_weight			,		threshold=quantization_config.llm_inta_threshold			,		)
                        snake_case								= True
                    else:
                        if (
                            quantization_config.llm_inta_skip_modules is not None
                            and name in quantization_config.llm_inta_skip_modules
                        ):
                            pass
                        else:
                            snake_case								= bnb.nn.Linearabit(
                                A__			,		A__			,		module.bias is not None			,		quantization_config.bnb_abit_compute_dtype			,		compress_statistics=quantization_config.bnb_abit_use_double_quant			,		quant_type=quantization_config.bnb_abit_quant_type			,		)
                            snake_case								= True
                    # Store the module class in case we need to transpose the weight later
                    snake_case								= type(A__		)
                    # Force requires grad to False to avoid unexpected errors
                    model._modules[name].requires_grad_(A__		)
        if len(list(module.children()		)		) > 0:
            snake_case					,							snake_case								= _replace_with_bnb_linear(
                A__			,		A__			,		A__			,		A__			,		has_been_replaced=A__			,		)
        # Remove the last key for recursion
        current_key_name.pop(-1		)
    return model, has_been_replaced



def 						lowercase_      (       A__			,		A__=None			,		A__=None			,		A__=None		)							->							List[str]:
    """simple docstring"""

    snake_case								= ["lm_head"] if modules_to_not_convert is None else modules_to_not_convert
    snake_case					,							snake_case								= _replace_with_bnb_linear(
        A__			,		A__			,		A__			,		A__		)

    if not has_been_replaced:
        logger.warning(
            "You are loading your model in 8bit or 4bit but no linear modules were found in your model."
            " Please double check your model architecture, or submit an issue on github if you think this is"
            " a bug."		)

    return model



def 						lowercase_      (       *A__			,		**A__		)							->							Optional[int]:
    """simple docstring"""

    warnings.warn(
        "`replace_8bit_linear` will be deprecated in a future version, please use `replace_with_bnb_linear` instead"			,		A__			,		)
    return replace_with_bnb_linear(*A__			,		**A__		)



def 						lowercase_      (       *A__			,		**A__		)							->							Any:
    """simple docstring"""

    warnings.warn(
        "`set_module_8bit_tensor_to_device` will be deprecated in a future version, please use `set_module_quantized_tensor_to_device` instead"			,		A__			,		)
    return set_module_quantized_tensor_to_device(*A__			,		**A__		)





def 						lowercase_      (       A__		)							->							Union[str, Any]:
    """simple docstring"""

    snake_case								= deepcopy(A__		)  # this has 0 cost since it is done inside `init_empty_weights` context manager`
    tied_model.tie_weights()

    snake_case								= find_tied_parameters(A__		)
    # For compatibility with Accelerate < 0.18
    if isinstance(A__			,		A__		):
        snake_case								= sum(list(tied_params.values()		)			,		[]		) + list(tied_params.keys()		)
    else:
        snake_case								= sum(A__			,		[]		)
    snake_case								= len(A__		) > 0

    # Check if it is a base model
    snake_case								= not hasattr(A__			,		model.base_model_prefix		)

    # Ignore this for base models (BertModel, GPT2Model, etc.)
    if (not has_tied_params) and is_base_model:
        return []

    # otherwise they have an attached head
    snake_case								= list(model.named_children()		)
    snake_case								= [list_modules[-1][0]]

    # add last module together with tied weights
    snake_case								= set(A__		) - set(A__		)
    snake_case								= list(set(A__		)		) + list(A__		)

    # remove ".weight" from the keys
    snake_case								= [".weight", ".bias"]
    snake_case								= []
    for name in list_untouched:
        for name_to_remove in names_to_remove:
            if name_to_remove in name:
                snake_case								= name.replace(A__			,		""		)
        filtered_module_names.append(A__		)

    return filtered_module_names

