
"""simple docstring"""




# 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     __magic_name__      ( __snake_case							:			int	,					__snake_case							:			str	,					__snake_case							:			int=0    )					->							Optional[int]:
       # Format the message.
       if name is None:
              lowercase       :							Union[str, Any]									=				None
       else:
              lowercase       :							Any									=				"." * max(0	,					spaces - 2    ) + "# {:" + str(50 - spaces    ) + "s}"
              lowercase       :							Dict									=				fmt.format(__snake_case    )

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


def     __magic_name__      ( __snake_case							:			Union[str, Any]	,					__snake_case							:			str	,					__snake_case							:			List[str]	,					__snake_case							:			List[str]	,					__snake_case							:			str    )					->							Any:
       # Permutes layout of param tensor to [num_splits * num_heads * hidden_size, :]
       # for compatibility with later versions of NVIDIA Megatron-LM.
       # The inverse operation is performed inside Megatron-LM to read checkpoints:
       # https://github.com/NVIDIA/Megatron-LM/blob/v2.4/megatron/checkpointing.py#L209
       # If param is the weight tensor of the self-attention block, the returned tensor
       # will have to be transposed one more time to be read by HuggingFace GPT2.
       lowercase       :							Optional[int]									=				param.size()
       if checkpoint_version == 1.0:
              # version 1.0 stores [num_heads * hidden_size * num_splits, :]
              lowercase       :							Optional[int]									=				(num_heads, hidden_size, num_splits) + input_shape[1:]
              lowercase       :							List[str]									=				param.view(*__snake_case    )
              lowercase       :							int									=				param.transpose(0	,					2    )
              lowercase       :							List[Any]									=				param.transpose(1	,					2    ).contiguous()
       elif checkpoint_version >= 2.0:
              # other versions store [num_heads * num_splits * hidden_size, :]
              lowercase       :							Union[str, Any]									=				(num_heads, num_splits, hidden_size) + input_shape[1:]
              lowercase       :							str									=				param.view(*__snake_case    )
              lowercase       :							Optional[int]									=				param.transpose(0	,					1    ).contiguous()
       lowercase       :							Union[str, Any]									=				param.view(*__snake_case    )
       return param


def     __magic_name__      ( __snake_case							:			Tuple	,					__snake_case							:			Union[str, Any]	,					__snake_case							:			Optional[int]    )					->							Optional[Any]:
       # The converted output model.
       lowercase       :							Tuple									=				{}

       # old versions did not store training args
       lowercase       :							Dict									=				input_state_dict.get("args"	,					__snake_case    )
       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))

              lowercase       :							str									=				ds_args.padded_vocab_size
              lowercase       :							List[Any]									=				ds_args.max_position_embeddings
              lowercase       :							int									=				ds_args.hidden_size
              lowercase       :							Union[str, Any]									=				ds_args.num_layers
              lowercase       :							int									=				ds_args.num_attention_heads
              lowercase       :							Union[str, Any]									=				ds_args.ffn_hidden_size
              # pprint(config)

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

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

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

       # The position embeddings.
       lowercase       :							Dict									=				embeddings["position_embeddings"]["weight"]
       # Read the causal mask dimension (seqlen). [max_sequence_length, hidden_size]
       lowercase       :							Tuple									=				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.
       lowercase       :							Optional[Any]									=				pos_embeddings

       # The transformer.
       lowercase       :							Optional[int]									=				lm["transformer"] if "transformer" in lm.keys() else lm["encoder"]

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

       # The simple map of names for "automated" rules.
       lowercase       :							Any									=				{
           "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.
              lowercase       :							str									=				layer_re.match(__snake_case    )

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

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

              # The name of the layer.
              lowercase       :							Union[str, Any]									=				f"""transformer.h.{layer_idx}"""

              # For layernorm(s), simply store the layer norm.
              if op_name.endswith("layernorm"    ):
                     lowercase       :							int									=				"ln_1" if op_name.startswith("input"    ) else "ln_2"
                     lowercase       :							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.
                     lowercase       :							Optional[Any]									=				torch.tril(torch.ones((n_positions, n_positions)	,					dtype=torch.floataa    )    ).view(
                         1	,					1	,					__snake_case	,					__snake_case    )
                     lowercase       :							Dict									=				causal_mask

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

                     lowercase       :							List[str]									=				fix_query_key_value_ordering(__snake_case	,					__snake_case	,					3	,					__snake_case	,					__snake_case    )
                     # Megatron stores (3*D) x D but transformers-GPT2 expects D x 3*D.
                     lowercase       :							Union[str, Any]									=				out_val.transpose(0	,					1    ).contiguous()
                     # Store.
                     lowercase       :							str									=				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":
                     lowercase       :							Tuple									=				fix_query_key_value_ordering(__snake_case	,					__snake_case	,					3	,					__snake_case	,					__snake_case    )
                     # Store. No change of shape.
                     lowercase       :							Union[str, Any]									=				out_val

              # Transpose the weights.
              elif weight_or_bias == "weight":
                     lowercase       :							Any									=				megatron_to_transformers[op_name]
                     lowercase       :							int									=				val.transpose(0	,					1    )

              # Copy the bias.
              elif weight_or_bias == "bias":
                     lowercase       :							str									=				megatron_to_transformers[op_name]
                     lowercase       :							Optional[Any]									=				val

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

       # The final layernorm.
       lowercase       :							Dict									=				transformer["final_layernorm.weight"]
       lowercase       :							Union[str, Any]									=				transformer["final_layernorm.bias"]

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

       # It should be done!
       return output_state_dict


def     __magic_name__      ( )					->							Optional[Any]:
       # Create the argument parser.
       lowercase       :							Dict									=				argparse.ArgumentParser()
       parser.add_argument("--print-checkpoint-structure"	,					action="store_true"    )
       parser.add_argument(
           "path_to_checkpoint"	,					type=__snake_case	,					help="Path to the checkpoint file (.zip archive or direct .pt file)"	,					)
       parser.add_argument(
           "--config_file"	,					default=""	,					type=__snake_case	,					help="An optional config json file describing the pre-trained model."	,					)
       lowercase       :							List[str]									=				parser.parse_args()

       # Extract the basename.
       lowercase       :							str									=				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:
                            lowercase       :							List[str]									=				torch.load(__snake_case	,					map_location="cpu"    )
       else:
              lowercase       :							Optional[int]									=				torch.load(args.path_to_checkpoint	,					map_location="cpu"    )

       lowercase       :							Optional[int]									=				input_state_dict.get("args"	,					__snake_case    )

       # 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:
                            lowercase       :							Tuple									=				"gelu_fast"
                     elif ds_args.openai_gelu:
                            lowercase       :							List[Any]									=				"gelu_new"
                     else:
                            lowercase       :							List[str]									=				"gelu"
              else:
                     # in the very early days this used to be "gelu_new"
                     lowercase       :							Any									=				"gelu_new"

              # Spell out all parameters in case the defaults change.
              lowercase       :							List[str]									=				GPTaConfig(
                  vocab_size=5_0257	,					n_positions=1024	,					n_embd=1024	,					n_layer=24	,					n_head=16	,					n_inner=4096	,					activation_function=__snake_case	,					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=__snake_case	,					summary_activation=__snake_case	,					summary_proj_to_labels=__snake_case	,					summary_first_dropout=0.1	,					scale_attn_weights=__snake_case	,					use_cache=__snake_case	,					bos_token_id=5_0256	,					eos_token_id=5_0256	,					)
       else:
              lowercase       :							Any									=				GPTaConfig.from_json_file(args.config_file    )

       lowercase       :							List[str]									=				["GPT2LMHeadModel"]

       # Convert.
       print("Converting"    )
       lowercase       :							List[str]									=				convert_megatron_checkpoint(__snake_case	,					__snake_case	,					__snake_case    )

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

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

       lowercase       :							int									=				AutoTokenizer.from_pretrained(__snake_case    )
       lowercase       :							Optional[Any]									=				type(__snake_case    ).__name__
       lowercase       :							Union[str, Any]									=				tokenizer_class

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

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

       # Store the state_dict to file.
       lowercase       :							Optional[Any]									=				os.path.join(__snake_case	,					"pytorch_model.bin"    )
       print(f"""Saving checkpoint to \"{output_checkpoint_file}\""""    )
       torch.save(__snake_case	,					__snake_case    )


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

if __name__ == "__main__":
   main()

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




"""simple docstring"""




import unittest

import numpy as np
import torch

from diffusers import KarrasVePipeline, KarrasVeScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()

class 						a__		( unittest.TestCase		):
   @property
   def      __magic_name__     (     self							):
          torch.manual_seed(0							)
          lowercase       :							str									=				UNetaDModel(
              block_out_channels=(32, 64)      ,       layers_per_block=2      ,       sample_size=32      ,       in_channels=3      ,       out_channels=3      ,       down_block_types=("DownBlock2D", "AttnDownBlock2D")      ,       up_block_types=("AttnUpBlock2D", "UpBlock2D")      ,       )
          return model

   def      __magic_name__     (     self							):
          lowercase       :							List[str]									=				self.dummy_uncond_unet
          lowercase       :							Dict									=				KarrasVeScheduler()

          lowercase       :							Any									=				KarrasVePipeline(unet=_a      ,       scheduler=_a							)
          pipe.to(_a							)
          pipe.set_progress_bar_config(disable=_a							)

          lowercase       :							Tuple									=				torch.manual_seed(0							)
          lowercase       :							Tuple									=				pipe(num_inference_steps=2      ,       generator=_a      ,       output_type="numpy"							).images

          lowercase       :							int									=				torch.manual_seed(0							)
          lowercase       :							Tuple									=				pipe(num_inference_steps=2      ,       generator=_a      ,       output_type="numpy"      ,       return_dict=_a							)[0]

          lowercase       :							Any									=				image[0, -3:, -3:, -1]
          lowercase       :							Dict									=				image_from_tuple[0, -3:, -3:, -1]

          assert image.shape == (1, 32, 32, 3)
          lowercase       :							Tuple									=				np.array([0.0, 1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0]							)

          assert np.abs(image_slice.flatten() - expected_slice							).max() < 1E-2
          assert np.abs(image_from_tuple_slice.flatten() - expected_slice							).max() < 1E-2


@slow
@require_torch
class 						a__		( unittest.TestCase		):
   def      __magic_name__     (     self							):
          lowercase       :							Tuple									=				"google/ncsnpp-celebahq-256"
          lowercase       :							str									=				UNetaDModel.from_pretrained(_a							)
          lowercase       :							Optional[Any]									=				KarrasVeScheduler()

          lowercase       :							Union[str, Any]									=				KarrasVePipeline(unet=_a      ,       scheduler=_a							)
          pipe.to(_a							)
          pipe.set_progress_bar_config(disable=_a							)

          lowercase       :							Tuple									=				torch.manual_seed(0							)
          lowercase       :							Dict									=				pipe(num_inference_steps=20      ,       generator=_a      ,       output_type="numpy"							).images

          lowercase       :							Optional[Any]									=				image[0, -3:, -3:, -1]
          assert image.shape == (1, 256, 256, 3)
          lowercase       :							Optional[Any]									=				np.array([0.5_7_8, 0.5_8_1_1, 0.5_9_2_4, 0.5_8_0_9, 0.5_8_7, 0.5_8_8_6, 0.5_8_6_1, 0.5_8_0_2, 0.5_8_6]							)

          assert np.abs(image_slice.flatten() - expected_slice							).max() < 1E-2


