
"""simple docstring"""
import argparse
import gc
import json
import os
import shutil
import warnings

import torch

from transformers import LlamaConfig, LlamaForCausalLM, LlamaTokenizer


try:
      from transformers import LlamaTokenizerFast
except ImportError as e:
      warnings.warn(e)
      warnings.warn(
          '''The converted tokenizer will be the `slow` tokenizer. To use the fast, update your `tokenizers` library and re-run the tokenizer conversion'''
      )
      lowercase_							:    Dict         =						None
lowercase_							:    Optional[Any]         =						{
    '''7B''': 1_10_08,
    '''13B''': 1_38_24,
    '''30B''': 1_79_20,
    '''65B''': 2_20_16,
    '''70B''': 2_86_72,
}
lowercase_							:    Any         =						{
    '''7B''': 1,
    '''7Bf''': 1,
    '''13B''': 2,
    '''13Bf''': 2,
    '''30B''': 4,
    '''65B''': 8,
    '''70B''': 8,
    '''70Bf''': 8,
}





def  _lowerCAmelCase     (  lowerCamelCase__				:				Union[str, Any],    lowerCamelCase__				:				Optional[Any]=1,    lowerCamelCase__				:				Any=2_5_6					)	->     Optional[int]:
  return multiple_of * ((int(ffn_dim_multiplier * int(8 * n / 3					)					) + multiple_of - 1) // multiple_of)





def  _lowerCAmelCase     (  lowerCamelCase__				:				Union[str, Any]					)	->     List[str]:
  with open(lowerCamelCase__,    "r"					) as f:
    return json.load(lowerCamelCase__					)





def  _lowerCAmelCase     (  lowerCamelCase__				:				str,    lowerCamelCase__				:				List[Any]					)	->     Union[str, Any]:
  with open(lowerCamelCase__,    "w"					) as f:
    json.dump(lowerCamelCase__,    lowerCamelCase__					)





def  _lowerCAmelCase     (  lowerCamelCase__				:				List[Any],    lowerCamelCase__				:				int,    lowerCamelCase__				:				Tuple,    lowerCamelCase__				:				Tuple=True					)	->     Any:
  os.makedirs(lowerCamelCase__,    exist_ok=lowerCamelCase__					)
  _SCREAMING_SNAKE_CASE :					Tuple				     =			os.path.join(lowerCamelCase__,    "tmp"					)
  os.makedirs(lowerCamelCase__,    exist_ok=lowerCamelCase__					)

  _SCREAMING_SNAKE_CASE :					Dict				     =			read_json(os.path.join(lowerCamelCase__,    "params.json"					)					)
  _SCREAMING_SNAKE_CASE :					str				     =			NUM_SHARDS[model_size]
  _SCREAMING_SNAKE_CASE :					int				     =			params['''n_layers''']
  _SCREAMING_SNAKE_CASE :					List[Any]				     =			params['''n_heads''']
  _SCREAMING_SNAKE_CASE :					Any				     =			n_heads // num_shards
  _SCREAMING_SNAKE_CASE :					int				     =			params['''dim''']
  _SCREAMING_SNAKE_CASE :					Tuple				     =			dim // n_heads
  _SCREAMING_SNAKE_CASE :					int				     =			10000.0
  _SCREAMING_SNAKE_CASE :					List[Any]				     =			1.0 / (base ** (torch.arange(0,    lowerCamelCase__,    2					).float() / dims_per_head))

  if "n_kv_heads" in params:
    _SCREAMING_SNAKE_CASE :					Tuple				     =			params['''n_kv_heads''']  # for GQA / MQA
    _SCREAMING_SNAKE_CASE :					Any				     =			n_heads_per_shard // num_key_value_heads
    _SCREAMING_SNAKE_CASE :					Dict				     =			dim // num_key_value_heads
  else:  # compatibility with other checkpoints
    _SCREAMING_SNAKE_CASE :					Dict				     =			n_heads
    _SCREAMING_SNAKE_CASE :					Any				     =			n_heads_per_shard
    _SCREAMING_SNAKE_CASE :					List[str]				     =			dim

  # permute for sliced rotary
  def permute(lowerCamelCase__				:				Tuple,    lowerCamelCase__				:				Optional[Any]=n_heads,    lowerCamelCase__				:				int=dim,    lowerCamelCase__				:				Optional[int]=dim					):
    return w.view(lowerCamelCase__,    dima // n_heads // 2,    2,    lowerCamelCase__					).transpose(1,    2					).reshape(lowerCamelCase__,    lowerCamelCase__					)

  print(f'''Fetching all parameters from the checkpoint at {input_base_path}.'''					)
  # Load weights
  if model_size == "7B":
    # Not sharded
    # (The sharded implementation would also work, but this is simpler.)
    _SCREAMING_SNAKE_CASE :					Dict				     =			torch.load(os.path.join(lowerCamelCase__,    "consolidated.00.pth"					),    map_location="cpu"					)
  else:
    # Sharded
    _SCREAMING_SNAKE_CASE :					Optional[int]				     =			[
        torch.load(os.path.join(lowerCamelCase__,    f'''consolidated.{i:02d}.pth'''					),    map_location="cpu"					)
        for i in range(lowerCamelCase__					)
    ]
  _SCREAMING_SNAKE_CASE :					Dict				     =			0
  _SCREAMING_SNAKE_CASE :					Optional[Any]				     =			{'''weight_map''': {}}
  for layer_i in range(lowerCamelCase__					):
    _SCREAMING_SNAKE_CASE :					List[Any]				     =			f'''pytorch_model-{layer_i + 1}-of-{n_layers + 1}.bin'''
    if model_size == "7B":
      # Unsharded
      _SCREAMING_SNAKE_CASE :					int				     =			{
          f'''model.layers.{layer_i}.self_attn.q_proj.weight''': permute(
              loaded[f'''layers.{layer_i}.attention.wq.weight''']					),
          f'''model.layers.{layer_i}.self_attn.k_proj.weight''': permute(
              loaded[f'''layers.{layer_i}.attention.wk.weight''']					),
          f'''model.layers.{layer_i}.self_attn.v_proj.weight''': loaded[f'''layers.{layer_i}.attention.wv.weight'''],
          f'''model.layers.{layer_i}.self_attn.o_proj.weight''': loaded[f'''layers.{layer_i}.attention.wo.weight'''],
          f'''model.layers.{layer_i}.mlp.gate_proj.weight''': loaded[f'''layers.{layer_i}.feed_forward.w1.weight'''],
          f'''model.layers.{layer_i}.mlp.down_proj.weight''': loaded[f'''layers.{layer_i}.feed_forward.w2.weight'''],
          f'''model.layers.{layer_i}.mlp.up_proj.weight''': loaded[f'''layers.{layer_i}.feed_forward.w3.weight'''],
          f'''model.layers.{layer_i}.input_layernorm.weight''': loaded[f'''layers.{layer_i}.attention_norm.weight'''],
          f'''model.layers.{layer_i}.post_attention_layernorm.weight''': loaded[f'''layers.{layer_i}.ffn_norm.weight'''],
      }
    else:
      # Sharded
      # Note that attention.w{q,k,v,o}, feed_fordward.w[1,2,3], attention_norm.weight and ffn_norm.weight share
      # the same storage object, saving attention_norm and ffn_norm will save other weights too, which is
      # redundant as other weights will be stitched from multiple shards. To avoid that, they are cloned.

      _SCREAMING_SNAKE_CASE :					Tuple				     =			{
          f'''model.layers.{layer_i}.input_layernorm.weight''': loaded[0][
              f'''layers.{layer_i}.attention_norm.weight'''
          ].clone(),
          f'''model.layers.{layer_i}.post_attention_layernorm.weight''': loaded[0][
              f'''layers.{layer_i}.ffn_norm.weight'''
          ].clone(),
      }
      _SCREAMING_SNAKE_CASE :					Tuple				     =			permute(
          torch.cat(
              [
                  loaded[i][f'''layers.{layer_i}.attention.wq.weight'''].view(lowerCamelCase__,    lowerCamelCase__,    lowerCamelCase__					)
                  for i in range(lowerCamelCase__					)
              ],    dim=0,    ).reshape(lowerCamelCase__,    lowerCamelCase__					)					)
      _SCREAMING_SNAKE_CASE :					Dict				     =			permute(
          torch.cat(
              [
                  loaded[i][f'''layers.{layer_i}.attention.wk.weight'''].view(
                      lowerCamelCase__,    lowerCamelCase__,    lowerCamelCase__					)
                  for i in range(lowerCamelCase__					)
              ],    dim=0,    ).reshape(lowerCamelCase__,    lowerCamelCase__					),    lowerCamelCase__,    lowerCamelCase__,    lowerCamelCase__,    )
      _SCREAMING_SNAKE_CASE :					Optional[int]				     =			torch.cat(
          [
              loaded[i][f'''layers.{layer_i}.attention.wv.weight'''].view(
                  lowerCamelCase__,    lowerCamelCase__,    lowerCamelCase__					)
              for i in range(lowerCamelCase__					)
          ],    dim=0,    ).reshape(lowerCamelCase__,    lowerCamelCase__					)

      _SCREAMING_SNAKE_CASE :					Any				     =			torch.cat(
          [loaded[i][f'''layers.{layer_i}.attention.wo.weight'''] for i in range(lowerCamelCase__					)],    dim=1					)
      _SCREAMING_SNAKE_CASE :					Union[str, Any]				     =			torch.cat(
          [loaded[i][f'''layers.{layer_i}.feed_forward.w1.weight'''] for i in range(lowerCamelCase__					)],    dim=0					)
      _SCREAMING_SNAKE_CASE :					Dict				     =			torch.cat(
          [loaded[i][f'''layers.{layer_i}.feed_forward.w2.weight'''] for i in range(lowerCamelCase__					)],    dim=1					)
      _SCREAMING_SNAKE_CASE :					List[Any]				     =			torch.cat(
          [loaded[i][f'''layers.{layer_i}.feed_forward.w3.weight'''] for i in range(lowerCamelCase__					)],    dim=0					)

    _SCREAMING_SNAKE_CASE :					Any				     =			inv_freq
    for k, v in state_dict.items():
      _SCREAMING_SNAKE_CASE :					int				     =			filename
      param_count += v.numel()
    torch.save(lowerCamelCase__,    os.path.join(lowerCamelCase__,    lowerCamelCase__					)					)

  _SCREAMING_SNAKE_CASE :					Optional[int]				     =			f'''pytorch_model-{n_layers + 1}-of-{n_layers + 1}.bin'''
  if model_size == "7B":
    # Unsharded
    _SCREAMING_SNAKE_CASE :					List[Any]				     =			{
        '''model.embed_tokens.weight''': loaded['''tok_embeddings.weight'''],
        '''model.norm.weight''': loaded['''norm.weight'''],
        '''lm_head.weight''': loaded['''output.weight'''],
    }
  else:
    _SCREAMING_SNAKE_CASE :					Dict				     =			{
        '''model.norm.weight''': loaded[0]['''norm.weight'''],
        '''model.embed_tokens.weight''': torch.cat(
            [loaded[i]["tok_embeddings.weight"] for i in range(lowerCamelCase__					)],    dim=1					),
        '''lm_head.weight''': torch.cat([loaded[i]["output.weight"] for i in range(lowerCamelCase__					)],    dim=0					),
    }

  for k, v in state_dict.items():
    _SCREAMING_SNAKE_CASE :					List[str]				     =			filename
    param_count += v.numel()
  torch.save(lowerCamelCase__,    os.path.join(lowerCamelCase__,    lowerCamelCase__					)					)

  # Write configs
  _SCREAMING_SNAKE_CASE :					Tuple				     =			{'''total_size''': param_count * 2}
  write_json(lowerCamelCase__,    os.path.join(lowerCamelCase__,    "pytorch_model.bin.index.json"					)					)
  _SCREAMING_SNAKE_CASE :					int				     =			params['''ffn_dim_multiplier'''] if '''ffn_dim_multiplier''' in params else 1
  _SCREAMING_SNAKE_CASE :					str				     =			params['''multiple_of'''] if '''multiple_of''' in params else 2_5_6
  _SCREAMING_SNAKE_CASE :					Any				     =			LlamaConfig(
      hidden_size=lowerCamelCase__,    intermediate_size=compute_intermediate_size(lowerCamelCase__,    lowerCamelCase__,    lowerCamelCase__					),    num_attention_heads=params["n_heads"],    num_hidden_layers=params["n_layers"],    rms_norm_eps=params["norm_eps"],    num_key_value_heads=lowerCamelCase__,    )
  config.save_pretrained(lowerCamelCase__					)

  # Make space so we can load the model properly now.
  del state_dict
  del loaded
  gc.collect()

  print("Loading the checkpoint in a Llama model."					)
  _SCREAMING_SNAKE_CASE :					int				     =			LlamaForCausalLM.from_pretrained(lowerCamelCase__,    torch_dtype=torch.floataa,    low_cpu_mem_usage=lowerCamelCase__					)
  # Avoid saving this as part of the config.
  del model.config._name_or_path

  print("Saving in the Transformers format."					)
  model.save_pretrained(lowerCamelCase__,    safe_serialization=lowerCamelCase__					)
  shutil.rmtree(lowerCamelCase__					)





def  _lowerCAmelCase     (  lowerCamelCase__				:				Union[str, Any],    lowerCamelCase__				:				Optional[int]					)	->     Any:
  _SCREAMING_SNAKE_CASE :					Optional[Any]				     =			LlamaTokenizer if LlamaTokenizerFast is None else LlamaTokenizerFast
  print(f'''Saving a {tokenizer_class.__name__} to {tokenizer_path}.'''					)
  _SCREAMING_SNAKE_CASE :					str				     =			tokenizer_class(lowerCamelCase__					)
  tokenizer.save_pretrained(lowerCamelCase__					)





def  _lowerCAmelCase     (  )	->     Tuple:
  _SCREAMING_SNAKE_CASE :					Optional[Any]				     =			argparse.ArgumentParser()
  parser.add_argument(
      "--input_dir",    help="Location of LLaMA weights, which contains tokenizer.model and model folders",    )
  parser.add_argument(
      "--model_size",    choices=["7B", "7Bf", "13B", "13Bf", "30B", "65B", "70B", "70Bf", "tokenizer_only"],    )
  parser.add_argument(
      "--output_dir",    help="Location to write HF model and tokenizer",    )
  parser.add_argument("--safe_serialization",    type=lowerCamelCase__,    help="Whether or not to save using `safetensors`."					)
  _SCREAMING_SNAKE_CASE :					List[Any]				     =			parser.parse_args()
  if args.model_size != "tokenizer_only":
    write_model(
        model_path=args.output_dir,    input_base_path=os.path.join(args.input_dir,    args.model_size					),    model_size=args.model_size,    safe_serialization=args.safe_serialization,    )
  _SCREAMING_SNAKE_CASE :					Dict				     =			os.path.join(args.input_dir,    "tokenizer.model"					)
  write_tokenizer(args.output_dir,    lowerCamelCase__					)


if __name__ == "__main__":
      main()




from pathlib import Path

import fire




def    lowerCAmelCase_  (		lowercase:				str	,       lowercase:				str	,       lowercase:				int		)     ->					int:







 '''simple docstring'''
 _UpperCamelCase:  Any    						=			Path(lowercase		)
 _UpperCamelCase:  int    						=			Path(lowercase		)
 dest_dir.mkdir(exist_ok=lowercase		)
 for path in src_dir.iterdir():
  _UpperCamelCase:  List[str]    						=			[x.rstrip() for x in list(path.open().readlines()		)][:n]
  _UpperCamelCase:  int    						=			dest_dir.joinpath(path.name		)
  print(lowercase		)
  dest_path.open('''w'''		).write('''\n'''.join(lowercase		)		)


if __name__ == "__main__":
  fire.Fire(minify)