


"""simple docstring"""



import argparse
import os
import re
import zipfile

import torch

from transformers import AutoTokenizer, GPTaConfig







def   __lowerCAmelCase						(						lowercase  :  Any						,							lowercase  :  Any						,							lowercase  :  Optional[int]=0 )	->						str:


    """simple docstring"""





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

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







def   __lowerCAmelCase						(						lowercase  :  Dict						,							lowercase  :  str						,							lowercase  :  str						,							lowercase  :  str						,							lowercase  :  Optional[int] )	->						Any:


    """simple docstring"""





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







def   __lowerCAmelCase						(						lowercase  :  Dict						,							lowercase  :  Optional[Any]						,							lowercase  :  int )	->						Union[str, Any]:


    """simple docstring"""





    snake_case    :      Union[str, Any]      					=    {}

    # old versions did not store training args
    snake_case    :      Dict      					=    input_state_dict.get("args"						,							lowercase )
    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    :      List[Any]      					=    ds_args.padded_vocab_size
        snake_case    :      Dict      					=    ds_args.max_position_embeddings
        snake_case    :      Optional[int]      					=    ds_args.hidden_size
        snake_case    :      Tuple      					=    ds_args.num_layers
        snake_case    :      Tuple      					=    ds_args.num_attention_heads
        snake_case    :      str      					=    ds_args.ffn_hidden_size
        # pprint(config)

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

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

    # The word embeddings.
    snake_case    :      List[Any]      					=    embeddings["word_embeddings"]["weight"]
    # Truncate the embedding table to vocab_size rows.
    snake_case    :      Tuple      					=    word_embeddings[: config.vocab_size, :]
    snake_case    :      Optional[int]      					=    word_embeddings

    # The position embeddings.
    snake_case    :      List[str]      					=    embeddings["position_embeddings"]["weight"]
    # Read the causal mask dimension (seqlen). [max_sequence_length, hidden_size]
    snake_case    :      Union[str, 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    :      Dict      					=    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    :      str      					=    re.compile(R"layers\.(\d+)\.([a-z0-9_.]+)\.([a-z]+)" )

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

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

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

        # The name of the layer.
        snake_case    :      List[str]      					=    F'transformer.h.{layer_idx}'

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

            # Insert a "dummy" tensor for masked_bias.
            snake_case    :      Any      					=    torch.tensor(-1e4						,							dtype=torch.floataa )
            snake_case    :      str      					=    masked_bias

            snake_case    :      Tuple      					=    fix_query_key_value_ordering(lowercase						,							lowercase						,							3						,							lowercase						,							lowercase )
            # 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    :      Tuple      					=    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    :      Tuple      					=    fix_query_key_value_ordering(lowercase						,							lowercase						,							3						,							lowercase						,							lowercase )
            # Store. No change of shape.
            snake_case    :      Optional[Any]      					=    out_val

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

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

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

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

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

    # It should be done!
    return output_state_dict







def   __lowerCAmelCase						(						)	->						List[Any]:


    """simple docstring"""





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

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

    snake_case    :      List[Any]      					=    input_state_dict.get("args"						,							lowercase )

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

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

    snake_case    :      Union[str, Any]      					=    ["GPT2LMHeadModel"]

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

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

    # Add tokenizer class info to config
    # see https://github.com/huggingface/transformers/issues/13906)
    if ds_args is not None:
        snake_case    :      Union[str, Any]      					=    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    :      List[str]      					=    "gpt2"

    snake_case    :      Any      					=    AutoTokenizer.from_pretrained(lowercase )
    snake_case    :      List[Any]      					=    type(lowercase ).__name__
    snake_case    :      Dict      					=    tokenizer_class

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

    # Save tokenizer based on args
    print(F'Adding {tokenizer_class} tokenizer files' )
    tokenizer.save_pretrained(lowercase )

    # Store the state_dict to file.
    snake_case    :      Tuple      					=    os.path.join(lowercase						,							"pytorch_model.bin" )
    print(F'Saving checkpoint to \"{output_checkpoint_file}\"' )
    torch.save(lowercase						,							lowercase )


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

if __name__ == "__main__":
  main()

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









"""simple docstring"""







from __future__ import annotations

from functools import lru_cache
from math import ceil

lowerCAmelCase__      				=   100

lowerCAmelCase__      				=   set(range(3, NUM_PRIMES, 2))
primes.add(2)
lowerCAmelCase__      				=   42

for prime in range(3, ceil(NUM_PRIMES**0.5), 2):
     if prime not in primes:
          continue
     primes.difference_update(set(range(prime * prime, NUM_PRIMES, prime)))
@lru_cache(maxsize=1_0_0				)
def 					a__			(				SCREAMING_SNAKE_CASE    :    int				):


  '''simple docstring'''







  if number_to_partition < 0:
    return set()
  elif number_to_partition == 0:
    return {1}

  lowerCAmelCase     :     set[int]								=							set()
  lowerCAmelCase     :     int
  lowerCAmelCase     :     int

  for prime in primes:
    if prime > number_to_partition:
      continue
    for sub in partition(number_to_partition - prime				):
      ret.add(sub * prime				)

  return ret
def 					a__			(				SCREAMING_SNAKE_CASE    :    int = 5_0_0_0				):


  '''simple docstring'''







  for number_to_partition in range(1				,      SCREAMING_SNAKE_CASE				):
    if len(partition(SCREAMING_SNAKE_CASE				)				) > number_unique_partitions:
      return number_to_partition
  return None


if __name__ == "__main__":
     print(F"{solution() = }")





