



"""simple docstring"""

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

lowercase__					:						str												=			{
    '''b0''': efficientnet.EfficientNetBa,
    '''b1''': efficientnet.EfficientNetBa,
    '''b2''': efficientnet.EfficientNetBa,
    '''b3''': efficientnet.EfficientNetBa,
    '''b4''': efficientnet.EfficientNetBa,
    '''b5''': efficientnet.EfficientNetBa,
    '''b6''': efficientnet.EfficientNetBa,
    '''b7''': efficientnet.EfficientNetBa,
}

lowercase__					:						List[str]												=			{
    '''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					):
    snake_case_				:  Union[str, Any]       =			EfficientNetConfig()
    snake_case_				:  List[str]       =			CONFIG_MAP[model_name]['''hidden_dim''']
    snake_case_				:  Tuple       =			CONFIG_MAP[model_name]['''width_coef''']
    snake_case_				:  str       =			CONFIG_MAP[model_name]['''depth_coef''']
    snake_case_				:  Dict       =			CONFIG_MAP[model_name]['''image_size''']
    snake_case_				:  Tuple       =			CONFIG_MAP[model_name]['''dropout_rate''']
    snake_case_				:  Any       =			CONFIG_MAP[model_name]['''dw_padding''']

    snake_case_				:  int       =			'''huggingface/label-files'''
    snake_case_				:  List[str]       =			'''imagenet-1k-id2label.json'''
    snake_case_				:  Union[str, Any]       =			1_000
    snake_case_				:  str       =			json.load(open(hf_hub_download(_a	,    _a	,    repo_type='''dataset'''					)	,    '''r'''					)					)
    snake_case_				:  Dict       =			{int(_a					): v for k, v in idalabel.items()}

    snake_case_				:  List[str]       =			idalabel
    snake_case_				:  Tuple       =			{v: k for k, v in idalabel.items()}
    return config




def __lowercase    (							):
    snake_case_				:  Optional[int]       =			'''http://images.cocodataset.org/val2017/000000039769.jpg'''
    snake_case_				:  Tuple       =			Image.open(requests.get(_a	,    stream=_a					).raw					)
    return im




def __lowercase    (							_a					):
    snake_case_				:  Optional[Any]       =			CONFIG_MAP[model_name]['''image_size''']
    snake_case_				:  List[str]       =			EfficientNetImageProcessor(
        size={'''height''': size, '''width''': size}	,    image_mean=[0.485, 0.456, 0.406]	,    image_std=[0.4785_3944, 0.473_2864, 0.4743_4163]	,    do_center_crop=_a	,    )
    return preprocessor




def __lowercase    (							_a					):
    snake_case_				:  Union[str, Any]       =			[v.split('''_'''					)[0].split('''block'''					)[1] for v in original_param_names if v.startswith('''block'''					)]
    snake_case_				:  int       =			sorted(set(_a					)					)
    snake_case_				:  Tuple       =			len(_a					)
    snake_case_				:  Union[str, Any]       =			{b: str(_a					) for b, i in zip(_a	,    range(_a					)					)}

    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:
        snake_case_				:  Any       =			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_				:  Dict       =			{}
    for item in rename_keys:
        if item[0] in original_param_names:
            snake_case_				:  str       =			'''efficientnet.''' + item[1]

    snake_case_				:  List[str]       =			'''classifier.weight'''
    snake_case_				:  List[str]       =			'''classifier.bias'''
    return key_mapping




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

        snake_case_				:  Tuple       =			key_mapping[key]
        if "_conv" in key and "kernel" in key:
            snake_case_				:  str       =			torch.from_numpy(_a					).permute(3	,    2	,    0	,    1					)
        elif "depthwise_kernel" in key:
            snake_case_				:  Union[str, Any]       =			torch.from_numpy(_a					).permute(2	,    3	,    0	,    1					)
        elif "kernel" in key:
            snake_case_				:  List[Any]       =			torch.from_numpy(np.transpose(_a					)					)
        else:
            snake_case_				:  List[str]       =			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					):
    snake_case_				:  Dict       =			model_classes[model_name](
        include_top=_a	,    weights='''imagenet'''	,    input_tensor=_a	,    input_shape=_a	,    pooling=_a	,    classes=1_000	,    classifier_activation='''softmax'''	,    )

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

    # Load HuggingFace model
    snake_case_				:  Tuple       =			get_efficientnet_config(_a					)
    snake_case_				:  Tuple       =			EfficientNetForImageClassification(_a					).eval()
    snake_case_				:  Dict       =			hf_model.state_dict()

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

    # Initialize preprocessor and preprocess input image
    snake_case_				:  Optional[Any]       =			convert_image_processor(_a					)
    snake_case_				:  List[Any]       =			preprocessor(images=prepare_img()	,    return_tensors='''pt'''					)

    # HF model inference
    hf_model.eval()
    with torch.no_grad():
        snake_case_				:  Union[str, Any]       =			hf_model(**_a					)
    snake_case_				:  List[str]       =			outputs.logits.detach().numpy()

    # Original model inference
    snake_case_				:  Union[str, Any]       =			False
    snake_case_				:  List[str]       =			CONFIG_MAP[model_name]['''image_size''']
    snake_case_				:  str       =			prepare_img().resize((image_size, image_size)	,    resample=PIL.Image.NEAREST					)
    snake_case_				:  Tuple       =			image.img_to_array(_a					)
    snake_case_				:  int       =			np.expand_dims(_a	,    axis=0					)
    snake_case_				:  Union[str, Any]       =			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_				:  Union[str, Any]       =			f"efficientnet-{model_name}"
        preprocessor.push_to_hub(_a					)
        hf_model.push_to_hub(_a					)


if __name__ == "__main__":
     lowercase__					:						Optional[Any]												=			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''')

     lowercase__					:						int												=			parser.parse_args()
     convert_efficientnet_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.save_model, args.push_to_hub)










"""simple docstring"""

import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
     import torch_xla.core.xla_model as xm
     import torch_xla.debug.metrics as met




class  _UpperCAmelCase (    lowerCAmelCase__):

 def __init__(  self	: List[Any]						,  *lowercase_	: List[Any]						,  lowercase_	: Optional[Any]=None						,  lowercase_	: Tuple=None						,  **lowercase_	: Union[str, Any]       ):
     super().__init__(*lowercase_						,  **lowercase_       )
     snake_case_				:  str       =			eval_examples
     snake_case_				:  Optional[Any]       =			post_process_function

 def 		_snake_case      (  self	: Union[str, Any]						,  lowercase_	: Optional[Dataset] = None						,  lowercase_	: str=None						,  lowercase_	: Optional[List[str]] = None						,  lowercase_	: str = "eval"						,  **lowercase_	: Dict						,  ):
     snake_case_				:  int       =			gen_kwargs.copy()
     snake_case_				:  Dict       =			(
         gen_kwargs['''max_length'''] if gen_kwargs.get('''max_length'''       ) is not None else self.args.generation_max_length
     )
     snake_case_				:  Tuple       =			(
         gen_kwargs['''num_beams'''] if gen_kwargs.get('''num_beams'''       ) is not None else self.args.generation_num_beams
     )
     snake_case_				:  Optional[Any]       =			gen_kwargs

     snake_case_				:  Optional[int]       =			self.eval_dataset if eval_dataset is None else eval_dataset
     snake_case_				:  Optional[Any]       =			self.get_eval_dataloader(lowercase_       )
     snake_case_				:  Dict       =			self.eval_examples if eval_examples is None else eval_examples

     # Temporarily disable metric computation, we will do it in the loop here.
     snake_case_				:  Optional[int]       =			self.compute_metrics
     snake_case_				:  str       =			None
     snake_case_				:  int       =			time.time()
     snake_case_				:  Any       =			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
     try:
         snake_case_				:  str       =			eval_loop(
             lowercase_						,  description='''Evaluation'''						,  prediction_loss_only=True if compute_metrics is None else None						,  ignore_keys=lowercase_						,  metric_key_prefix=lowercase_						,  )
     finally:
         snake_case_				:  Optional[int]       =			compute_metrics
     snake_case_				:  Dict       =			self.args.eval_batch_size * self.args.world_size
     if f"{metric_key_prefix}_jit_compilation_time" in output.metrics:
         start_time += output.metrics[f"{metric_key_prefix}_jit_compilation_time"]
     output.metrics.update(
         speed_metrics(
             lowercase_						,  lowercase_						,  num_samples=output.num_samples						,  num_steps=math.ceil(output.num_samples / total_batch_size       )						,  )       )

     if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
         # Only the main node write the results by default
         snake_case_				:  Tuple       =			self.post_process_function(lowercase_						,  lowercase_						,  lowercase_       )
         snake_case_				:  Optional[int]       =			self.compute_metrics(lowercase_       )

         # Prefix all keys with metric_key_prefix + '_'
         for key in list(metrics.keys()       ):
             if not key.startswith(f"{metric_key_prefix}_"       ):
                 snake_case_				:  int       =			metrics.pop(lowercase_       )

         metrics.update(output.metrics       )
     else:
         snake_case_				:  Tuple       =			output.metrics

     if self.args.should_log:
         # Only the main node log the results by default
         self.log(lowercase_       )

     if self.args.tpu_metrics_debug or self.args.debug:
         # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
         xm.master_print(met.metrics_report()       )

     snake_case_				:  str       =			self.callback_handler.on_evaluate(self.args						,  self.state						,  self.control						,  lowercase_       )
     return metrics






 def 		_snake_case      (  self	: Any						,  lowercase_	: int						,  lowercase_	: List[str]						,  lowercase_	: str=None						,  lowercase_	: str = "test"						,  **lowercase_	: Dict       ):
     snake_case_				:  Dict       =			gen_kwargs.copy()

     snake_case_				:  Dict       =			self.get_test_dataloader(lowercase_       )

     # Temporarily disable metric computation, we will do it in the loop here.
     snake_case_				:  Tuple       =			self.compute_metrics
     snake_case_				:  int       =			None
     snake_case_				:  List[Any]       =			time.time()
     snake_case_				:  int       =			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
     try:
         snake_case_				:  List[str]       =			eval_loop(
             lowercase_						,  description='''Prediction'''						,  prediction_loss_only=True if compute_metrics is None else None						,  ignore_keys=lowercase_						,  metric_key_prefix=lowercase_						,  )
     finally:
         snake_case_				:  str       =			compute_metrics

     snake_case_				:  int       =			self.args.eval_batch_size * self.args.world_size
     if f"{metric_key_prefix}_jit_compilation_time" in output.metrics:
         start_time += output.metrics[f"{metric_key_prefix}_jit_compilation_time"]
     output.metrics.update(
         speed_metrics(
             lowercase_						,  lowercase_						,  num_samples=output.num_samples						,  num_steps=math.ceil(output.num_samples / total_batch_size       )						,  )       )
     if self.post_process_function is None or self.compute_metrics is None:
         return output

     snake_case_				:  Tuple       =			self.post_process_function(lowercase_						,  lowercase_						,  lowercase_						,  '''predict'''       )
     snake_case_				:  Dict       =			self.compute_metrics(lowercase_       )

     # Prefix all keys with metric_key_prefix + '_'
     for key in list(metrics.keys()       ):
         if not key.startswith(f"{metric_key_prefix}_"       ):
             snake_case_				:  Union[str, Any]       =			metrics.pop(lowercase_       )
     metrics.update(output.metrics       )
     return PredictionOutput(predictions=predictions.predictions						,  label_ids=predictions.label_ids						,  metrics=lowercase_       )





