

# Copyright (c) 2021-, NVIDIA CORPORATION.  All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

####################################################################################################

#
# Note: If when running this conversion script you're getting an exception:
#     ModuleNotFoundError: No module named 'megatron.model.enums'
# you need to tell python where to find the clone of Megatron-LM, e.g.:
#
# cd /tmp
# git clone https://github.com/NVIDIA/Megatron-LM
# PYTHONPATH=/tmp/Megatron-LM python src/transformers/models/megatron_gpt2/convert_megatron_gpt2_checkpoint.py ...
#
# if you already have it cloned elsewhere, simply adjust the path to the existing path
#
# If the training was done using a Megatron-LM fork, e.g.,
# https://github.com/microsoft/Megatron-DeepSpeed/ then chances are that you need to have that one
# in your path, i.e., /path/to/Megatron-DeepSpeed/
#

import argparse
import os
import re
import zipfile

import torch

from transformers import AutoTokenizer, GPTaConfig







def       lowercase					(	__A :					Union[str, Any] ,    __A :					Any ,    __A :					Any=0		)				->			Any:





    '''simple docstring'''




    if name is None:
        snake_case							:		str        =		None
    else:
        snake_case							:		int        =		""".""" * max(0 ,    spaces - 2		) + """# {:""" + str(50 - spaces		) + """s}"""
        snake_case							:		Tuple        =		fmt.format(__A		)

    # Print and recurse (if needed).
    if isinstance(__A ,    __A		):
        if msg is not None:
            print(__A		)
        for k in val.keys():
            recursive_print(__A ,    val[k] ,    spaces + 2		)
    elif isinstance(__A ,    torch.Tensor		):
        print(__A ,    """:""" ,    val.size()		)
    else:
        print(__A ,    """:""" ,    __A		)







def       lowercase					(	__A :					Any ,    __A :					Dict ,    __A :					str ,    __A :					Tuple ,    __A :					Optional[int]		)				->			List[str]:





    '''simple docstring'''




    snake_case							:		int        =		param.size()
    if checkpoint_version == 1.0:
        # version 1.0 stores [num_heads * hidden_size * num_splits, :]
        snake_case							:		List[str]        =		(num_heads, hidden_size, num_splits) + input_shape[1:]
        snake_case							:		Optional[int]        =		param.view(*__A		)
        snake_case							:		Dict        =		param.transpose(0 ,    2		)
        snake_case							:		int        =		param.transpose(1 ,    2		).contiguous()
    elif checkpoint_version >= 2.0:
        # other versions store [num_heads * num_splits * hidden_size, :]
        snake_case							:		Tuple        =		(num_heads, num_splits, hidden_size) + input_shape[1:]
        snake_case							:		Union[str, Any]        =		param.view(*__A		)
        snake_case							:		Dict        =		param.transpose(0 ,    1		).contiguous()
    snake_case							:		Any        =		param.view(*__A		)
    return param







def       lowercase					(	__A :					str ,    __A :					Dict ,    __A :					str		)				->			List[Any]:





    '''simple docstring'''




    snake_case							:		Dict        =		{}

    # old versions did not store training args
    snake_case							:		int        =		input_state_dict.get("""args""" ,    __A		)
    if ds_args is not None:
        # do not make the user write a config file when the exact dimensions/sizes are already in the checkpoint
        # from pprint import pprint
        # pprint(vars(ds_args))

        snake_case							:		Dict        =		ds_args.padded_vocab_size
        snake_case							:		Dict        =		ds_args.max_position_embeddings
        snake_case							:		Optional[Any]        =		ds_args.hidden_size
        snake_case							:		List[str]        =		ds_args.num_layers
        snake_case							:		str        =		ds_args.num_attention_heads
        snake_case							:		Optional[Any]        =		ds_args.ffn_hidden_size
        # pprint(config)

    # The number of heads.
    snake_case							:		List[str]        =		config.n_head
    # The hidden_size per head.
    snake_case							:		List[Any]        =		config.n_embd // config.n_head
    # Megatron-LM checkpoint version
    if "checkpoint_version" in input_state_dict.keys():
        snake_case							:		List[Any]        =		input_state_dict["""checkpoint_version"""]
    else:
        snake_case							:		int        =		0.0

    # The model.
    snake_case							:		Optional[int]        =		input_state_dict["""model"""]
    # The language model.
    snake_case							:		Union[str, Any]        =		model["""language_model"""]
    # The embeddings.
    snake_case							:		List[str]        =		lm["""embedding"""]

    # The word embeddings.
    snake_case							:		Any        =		embeddings["""word_embeddings"""]["""weight"""]
    # Truncate the embedding table to vocab_size rows.
    snake_case							:		Union[str, Any]        =		word_embeddings[: config.vocab_size, :]
    snake_case							:		str        =		word_embeddings

    # The position embeddings.
    snake_case							:		Tuple        =		embeddings["""position_embeddings"""]["""weight"""]
    # Read the causal mask dimension (seqlen). [max_sequence_length, hidden_size]
    snake_case							:		Optional[Any]        =		pos_embeddings.size(0		)
    if n_positions != config.n_positions:
        raise ValueError(
            f"""pos_embeddings.max_sequence_length={n_positions} and config.n_positions={config.n_positions} don't match"""		)
    # Store the position embeddings.
    snake_case							:		Tuple        =		pos_embeddings

    # The transformer.
    snake_case							:		Union[str, Any]        =		lm["""transformer"""] if """transformer""" in lm.keys() else lm["""encoder"""]

    # The regex to extract layer names.
    snake_case							:		Optional[int]        =		re.compile(r"""layers\.(\d+)\.([a-z0-9_.]+)\.([a-z]+)"""		)

    # The simple map of names for "automated" rules.
    snake_case							:		str        =		{
        """attention.dense""": """.attn.c_proj.""",
        """self_attention.dense""": """.attn.c_proj.""",
        """mlp.dense_h_to_4h""": """.mlp.c_fc.""",
        """mlp.dense_4h_to_h""": """.mlp.c_proj.""",
    }

    # Extract the layers.
    for key, val in transformer.items():
        # Match the name.
        snake_case							:		Optional[Any]        =		layer_re.match(__A		)

        # Stop if that's not a layer
        if m is None:
            break

        # The index of the layer.
        snake_case							:		List[str]        =		int(m.group(1		)		)
        # The name of the operation.
        snake_case							:		Any        =		m.group(2		)
        # Is it a weight or a bias?
        snake_case							:		Any        =		m.group(3		)

        # The name of the layer.
        snake_case							:		Tuple        =		f"""transformer.h.{layer_idx}"""

        # For layernorm(s), simply store the layer norm.
        if op_name.endswith("""layernorm"""		):
            snake_case							:		Any        =		"""ln_1""" if op_name.startswith("""input"""		) else """ln_2"""
            snake_case							:		Any        =		val

        # Transpose the QKV matrix.
        elif (
            op_name == "attention.query_key_value" or op_name == "self_attention.query_key_value"
        ) and weight_or_bias == "weight":
            # Insert a tensor of 1x1xDxD bias.
            snake_case							:		Any        =		torch.tril(torch.ones((n_positions, n_positions) ,    dtype=torch.floataa		)		).view(
                1 ,    1 ,    __A ,    __A		)
            snake_case							:		List[str]        =		causal_mask

            # Insert a "dummy" tensor for masked_bias.
            snake_case							:		Optional[int]        =		torch.tensor(-1E4 ,    dtype=torch.floataa		)
            snake_case							:		Any        =		masked_bias

            snake_case							:		Optional[Any]        =		fix_query_key_value_ordering(__A ,    __A ,    3 ,    __A ,    __A		)
            # Megatron stores (3*D) x D but transformers-GPT2 expects D x 3*D.
            snake_case							:		int        =		out_val.transpose(0 ,    1		).contiguous()
            # Store.
            snake_case							:		List[Any]        =		out_val

        # Transpose the bias.
        elif (
            op_name == "attention.query_key_value" or op_name == "self_attention.query_key_value"
        ) and weight_or_bias == "bias":
            snake_case							:		List[str]        =		fix_query_key_value_ordering(__A ,    __A ,    3 ,    __A ,    __A		)
            # Store. No change of shape.
            snake_case							:		List[str]        =		out_val

        # Transpose the weights.
        elif weight_or_bias == "weight":
            snake_case							:		str        =		megatron_to_transformers[op_name]
            snake_case							:		Tuple        =		val.transpose(0 ,    1		)

        # Copy the bias.
        elif weight_or_bias == "bias":
            snake_case							:		List[Any]        =		megatron_to_transformers[op_name]
            snake_case							:		int        =		val

    # DEBUG.
    assert config.n_layer == layer_idx + 1

    # The final layernorm.
    snake_case							:		List[Any]        =		transformer["""final_layernorm.weight"""]
    snake_case							:		Any        =		transformer["""final_layernorm.bias"""]

    # For LM head, transformers' wants the matrix to weight embeddings.
    snake_case							:		List[Any]        =		word_embeddings

    # It should be done!
    return output_state_dict







def       lowercase					(	)				->			Any:





    '''simple docstring'''




    snake_case							:		List[Any]        =		argparse.ArgumentParser()
    parser.add_argument("""--print-checkpoint-structure""" ,    action="""store_true"""		)
    parser.add_argument(
        """path_to_checkpoint""" ,    type=__A ,    help="""Path to the checkpoint file (.zip archive or direct .pt file)""" ,    )
    parser.add_argument(
        """--config_file""" ,    default="""""" ,    type=__A ,    help="""An optional config json file describing the pre-trained model.""" ,    )
    snake_case							:		List[Any]        =		parser.parse_args()

    # Extract the basename.
    snake_case							:		Tuple        =		os.path.dirname(args.path_to_checkpoint		)

    # Load the model.
    # the .zip is very optional, let's keep it for backward compatibility
    print(f"""Extracting PyTorch state dictionary from {args.path_to_checkpoint}"""		)
    if args.path_to_checkpoint.endswith(""".zip"""		):
        with zipfile.ZipFile(args.path_to_checkpoint ,    """r"""		) as checkpoint:
            with checkpoint.open("""release/mp_rank_00/model_optim_rng.pt"""		) as pytorch_dict:
                snake_case							:		List[str]        =		torch.load(__A ,    map_location="""cpu"""		)
    else:
        snake_case							:		int        =		torch.load(args.path_to_checkpoint ,    map_location="""cpu"""		)

    snake_case							:		Dict        =		input_state_dict.get("""args""" ,    __A		)

    # Read the config, or default to the model released by NVIDIA.
    if args.config_file == "":
        if ds_args is not None:
            if ds_args.bias_gelu_fusion:
                snake_case							:		int        =		"""gelu_fast"""
            elif ds_args.openai_gelu:
                snake_case							:		Union[str, Any]        =		"""gelu_new"""
            else:
                snake_case							:		int        =		"""gelu"""
        else:
            # in the very early days this used to be "gelu_new"
            snake_case							:		str        =		"""gelu_new"""

        # Spell out all parameters in case the defaults change.
        snake_case							:		List[Any]        =		GPTaConfig(
            vocab_size=5_0257 ,    n_positions=1024 ,    n_embd=1024 ,    n_layer=24 ,    n_head=16 ,    n_inner=4096 ,    activation_function=__A ,    resid_pdrop=0.1 ,    embd_pdrop=0.1 ,    attn_pdrop=0.1 ,    layer_norm_epsilon=1E-5 ,    initializer_range=0.02 ,    summary_type="""cls_index""" ,    summary_use_proj=__A ,    summary_activation=__A ,    summary_proj_to_labels=__A ,    summary_first_dropout=0.1 ,    scale_attn_weights=__A ,    use_cache=__A ,    bos_token_id=5_0256 ,    eos_token_id=5_0256 ,    )
    else:
        snake_case							:		int        =		GPTaConfig.from_json_file(args.config_file		)

    snake_case							:		int        =		["""GPT2LMHeadModel"""]

    # Convert.
    print("""Converting"""		)
    snake_case							:		str        =		convert_megatron_checkpoint(__A ,    __A ,    __A		)

    # Print the structure of converted state dict.
    if args.print_checkpoint_structure:
        recursive_print(__A ,    __A		)

    # Add tokenizer class info to config
    # see https://github.com/huggingface/transformers/issues/13906)
    if ds_args is not None:
        snake_case							:		str        =		ds_args.tokenizer_type
        if tokenizer_type == "GPT2BPETokenizer":
            snake_case							:		Tuple        =		"""gpt2"""
        elif tokenizer_type == "PretrainedFromHF":
            snake_case							:		List[str]        =		ds_args.tokenizer_name_or_path
        else:
            raise ValueError(f"""Unrecognized tokenizer_type {tokenizer_type}"""		)
    else:
        snake_case							:		Any        =		"""gpt2"""

    snake_case							:		List[str]        =		AutoTokenizer.from_pretrained(__A		)
    snake_case							:		Optional[Any]        =		type(__A		).__name__
    snake_case							:		Tuple        =		tokenizer_class

    # Store the config to file.
    print("""Saving config"""		)
    config.save_pretrained(__A		)

    # Save tokenizer based on args
    print(f"""Adding {tokenizer_class} tokenizer files"""		)
    tokenizer.save_pretrained(__A		)

    # Store the state_dict to file.
    snake_case							:		List[str]        =		os.path.join(__A ,    """pytorch_model.bin"""		)
    print(f"""Saving checkpoint to \"{output_checkpoint_file}\""""		)
    torch.save(__A ,    __A		)


####################################################################################################

if __name__ == "__main__":
     main()

####################################################################################################




import argparse
import json
from pathlib import Path

import requests
import torch
from huggingface_hub import hf_hub_download
from PIL import Image

from transformers import (
    SwiftFormerConfig,
    SwiftFormerForImageClassification,
    ViTImageProcessor,
)
from transformers.utils import logging


logging.set_verbosity_info()
__a    :	Optional[Any]         =    logging.get_logger(__name__)

__a    :	Optional[Any]         =    torch.device("""cpu""")





def 							UpperCAmelCase		(       ):







				"""simple docstring"""



				__lowercase					     =				'''http://images.cocodataset.org/val2017/000000039769.jpg'''
				__lowercase					     =				Image.open(requests.get(lowercase				,     stream=lowercase				).raw				)
				return im





def 							UpperCAmelCase		(       lowercase				):







				"""simple docstring"""



				if swiftformer_name == "swiftformer_xs":
								return torch.tensor([-2.1703E00, 2.1107E00, -2.0811E00, 8.8685E-01, 2.4360E-01]				)

				elif swiftformer_name == "swiftformer_s":
								return torch.tensor([3.9636E-01, 2.3478E-01, -1.6963E00, -1.7381E00, -8.6337E-01]				)

				elif swiftformer_name == "swiftformer_l1":
								return torch.tensor([-4.2768E-01, -4.7429E-01, -1.0897E00, -1.0248E00, 3.5523E-02]				)

				elif swiftformer_name == "swiftformer_l3":
								return torch.tensor([-2.5330E-01, 2.4211E-01, -6.0185E-01, -8.2789E-01, -6.0446E-02]				)





def 							UpperCAmelCase		(       lowercase				,     lowercase				,     lowercase				):







				"""simple docstring"""



				__lowercase					     =				dct.pop(lowercase				)
				__lowercase					     =				val





def 							UpperCAmelCase		(       lowercase				):







				"""simple docstring"""



				__lowercase					     =				[]
				for k in state_dict.keys():
								__lowercase					     =				k
								if ".pwconv" in k:
												__lowercase					     =				k_new.replace('''.pwconv'''				,     '''.point_wise_conv'''				)
								if ".dwconv" in k:
												__lowercase					     =				k_new.replace('''.dwconv'''				,     '''.depth_wise_conv'''				)
								if ".Proj." in k:
												__lowercase					     =				k_new.replace('''.Proj.'''				,     '''.proj.'''				)
								if "patch_embed" in k_new:
												__lowercase					     =				k_new.replace('''patch_embed'''				,     '''swiftformer.patch_embed.patch_embedding'''				)
								if "network" in k_new:
												__lowercase					     =				k_new.split('''.'''				)
												if ls[2].isdigit():
																__lowercase					     =				'''swiftformer.encoder.network.''' + ls[1] + '''.blocks.''' + ls[2] + '''.''' + '''.'''.join(ls[3:]				)
												else:
																__lowercase					     =				k_new.replace('''network'''				,     '''swiftformer.encoder.network'''				)
								rename_keys.append((k, k_new)				)
				return rename_keys





@torch.no_grad()
def 							UpperCAmelCase		(       lowercase				,     lowercase				,     lowercase				):







				"""simple docstring"""



				__lowercase					     =				SwiftFormerConfig()

				# dataset (ImageNet-21k only or also fine-tuned on ImageNet 2012), patch_size and image_size
				__lowercase					     =				1000
				__lowercase					     =				'''huggingface/label-files'''
				__lowercase					     =				'''imagenet-1k-id2label.json'''
				__lowercase					     =				json.load(open(hf_hub_download(lowercase				,     lowercase				,     repo_type='''dataset'''				)				,     '''r'''				)				)
				__lowercase					     =				{int(lowercase				): v for k, v in idalabel.items()}
				__lowercase					     =				idalabel
				__lowercase					     =				{v: k for k, v in idalabel.items()}

				# size of the architecture
				if swiftformer_name == "swiftformer_xs":
								__lowercase					     =				[3, 3, 6, 4]
								__lowercase					     =				[48, 56, 112, 220]

				elif swiftformer_name == "swiftformer_s":
								__lowercase					     =				[3, 3, 9, 6]
								__lowercase					     =				[48, 64, 168, 224]

				elif swiftformer_name == "swiftformer_l1":
								__lowercase					     =				[4, 3, 10, 5]
								__lowercase					     =				[48, 96, 192, 384]

				elif swiftformer_name == "swiftformer_l3":
								__lowercase					     =				[4, 4, 12, 6]
								__lowercase					     =				[64, 128, 320, 512]

				# load state_dict of original model, remove and rename some keys
				if original_ckpt:
								if original_ckpt.startswith('''https'''				):
												__lowercase					     =				torch.hub.load_state_dict_from_url(lowercase				,     map_location='''cpu'''				,     check_hash=lowercase				)
								else:
												__lowercase					     =				torch.load(lowercase				,     map_location='''cpu'''				)
				__lowercase					     =				checkpoint

				__lowercase					     =				create_rename_keys(lowercase				)
				for rename_key_src, rename_key_dest in rename_keys:
								rename_key(lowercase				,     lowercase				,     lowercase				)

				# load HuggingFace model
				__lowercase					     =				SwiftFormerForImageClassification(lowercase				).eval()
				hf_model.load_state_dict(lowercase				)

				# prepare test inputs
				__lowercase					     =				prepare_img()
				__lowercase					     =				ViTImageProcessor.from_pretrained('''preprocessor_config'''				)
				__lowercase					     =				processor(images=lowercase				,     return_tensors='''pt'''				)

				# compare outputs from both models
				__lowercase					     =				get_expected_output(lowercase				)
				__lowercase					     =				hf_model(inputs['''pixel_values''']				).logits

				assert hf_logits.shape == torch.Size([1, 1000]				)
				assert torch.allclose(hf_logits[0, 0:5]				,     lowercase				,     atol=1E-3				)

				Path(lowercase				).mkdir(exist_ok=lowercase				)
				print(F"Saving model {swiftformer_name} to {pytorch_dump_folder_path}"				)
				hf_model.save_pretrained(lowercase				)


if __name__ == "__main__":
		__a    :	Optional[int]         =    argparse.ArgumentParser()
		# Required parameters
		parser.add_argument(
		    """--swiftformer_name""",
		    default="""swiftformer_xs""",
		    choices=["""swiftformer_xs""", """swiftformer_s""", """swiftformer_l1""", """swiftformer_l3"""],
		    type=str,
		    help="""Name of the SwiftFormer model you'd like to convert.""",
		)
		parser.add_argument(
		    """--pytorch_dump_folder_path""",
		    default="""./converted_outputs/""",
		    type=str,
		    help="""Path to the output PyTorch model directory.""",
		)
		parser.add_argument("""--original_ckpt""", default=None, type=str, help="""Path to the original model checkpoint.""")

		__a    :	Optional[int]         =    parser.parse_args()
		convert_swiftformer_checkpoint(args.swiftformer_name, args.pytorch_dump_folder_path, args.original_ckpt)