
from typing import List

from .keymap import KEYMAP, get_character






def  snake_case(    __magic_name__   )			->							str:





     '''simple docstring'''


     def decorator(__magic_name__   ):
          lowercase :       Tuple       =			getattr(lowercase_			,		'''handle_key'''			,		[]   )
          handle += [key]
          setattr(lowercase_			,		'''handle_key'''			,		lowercase_   )
          return func

     return decorator






def  snake_case(    *__magic_name__   )			->							str:





     '''simple docstring'''


     def decorator(__magic_name__   ):
          lowercase :       Union[str, Any]       =			getattr(lowercase_			,		'''handle_key'''			,		[]   )
          handle += keys
          setattr(lowercase_			,		'''handle_key'''			,		lowercase_   )
          return func

     return decorator
class 						_A (  __snake_case				):






  def __new__(  cls		:     str				,						_A		:     Tuple				,						_A		:     Dict				,						_A		:     List[str]				)		->   str:






       """simple docstring"""







       lowercase :       Tuple       =			super().__new__(cls				,						_A				,						_A				,						_A				)
       if not hasattr(_A				,						'''key_handler'''				):
            setattr(_A				,						'''key_handler'''				,						{}				)
       setattr(_A				,						'''handle_input'''				,						KeyHandler.handle_input				)

       for value in attrs.values():
            lowercase :       Tuple       =			getattr(_A				,						'''handle_key'''				,						[]				)
            for key in handled_keys:
                 lowercase :       Optional[int]       =			value
       return new_cls






  @staticmethod
  def  __a      (  cls		:     Dict				)		->   Tuple:






       """simple docstring"""







       lowercase :       Dict       =			get_character()
       if char != KEYMAP["undefined"]:
            lowercase :       Union[str, Any]       =			ord(_A				)
       lowercase :       Dict       =			cls.key_handler.get(_A				)
       if handler:
            lowercase :       Any       =			char
            return handler(cls				)
       else:
            return None






def  snake_case(    cls   )			->							List[str]:





     '''simple docstring'''


     return KeyHandler(cls.__name__			,		cls.__bases__			,		cls.__dict__.copy()   )





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'
 )
 lowerCAmelCase_						=			None
lowerCAmelCase_						=			{
    '7B': 1_10_08,
    '13B': 1_38_24,
    '30B': 1_79_20,
    '65B': 2_20_16,
    '70B': 2_86_72,
}
lowerCAmelCase_						=			{
    '7B': 1,
    '7Bf': 1,
    '13B': 2,
    '13Bf': 2,
    '30B': 4,
    '65B': 8,
    '70B': 8,
    '70Bf': 8,
}






def  snake_case(    __magic_name__			,		__magic_name__=1			,		__magic_name__=2_56   )			->							List[Any]:





     '''simple docstring'''


     return multiple_of * ((int(ffn_dim_multiplier * int(8 * n / 3   )   ) + multiple_of - 1) // multiple_of)






def  snake_case(    __magic_name__   )			->							Union[str, Any]:





     '''simple docstring'''


     with open(__magic_name__			,		'''r'''   ) as f:
          return json.load(__magic_name__   )






def  snake_case(    __magic_name__			,		__magic_name__   )			->							Optional[int]:





     '''simple docstring'''


     with open(__magic_name__			,		'''w'''   ) as f:
          json.dump(__magic_name__			,		__magic_name__   )






def  snake_case(    __magic_name__			,		__magic_name__			,		__magic_name__			,		__magic_name__=True   )			->							Tuple:





     '''simple docstring'''


     os.makedirs(__magic_name__			,		exist_ok=__magic_name__   )
     lowercase :       List[Any]       =			os.path.join(__magic_name__			,		'''tmp'''   )
     os.makedirs(__magic_name__			,		exist_ok=__magic_name__   )

     lowercase :       List[Any]       =			read_json(os.path.join(__magic_name__			,		'''params.json'''   )   )
     lowercase :       int       =			NUM_SHARDS[model_size]
     lowercase :       str       =			params['''n_layers''']
     lowercase :       Optional[int]       =			params['''n_heads''']
     lowercase :       str       =			n_heads // num_shards
     lowercase :       Dict       =			params['''dim''']
     lowercase :       int       =			dim // n_heads
     lowercase :       List[str]       =			1_0_0_0_0.0
     lowercase :       Optional[Any]       =			1.0 / (base ** (torch.arange(0			,		__magic_name__			,		2   ).float() / dims_per_head))

     if "n_kv_heads" in params:
          lowercase :       Optional[Any]       =			params['''n_kv_heads''']  # for GQA / MQA
          lowercase :       Union[str, Any]       =			n_heads_per_shard // num_key_value_heads
          lowercase :       Any       =			dim // num_key_value_heads
     else:  # compatibility with other checkpoints
          lowercase :       Optional[Any]       =			n_heads
          lowercase :       str       =			n_heads_per_shard
          lowercase :       Any       =			dim

     # permute for sliced rotary
     def permute(__magic_name__			,		__magic_name__=n_heads			,		__magic_name__=dim			,		__magic_name__=dim   ):
          return w.view(__magic_name__			,		dima // n_heads // 2			,		2			,		__magic_name__   ).transpose(1			,		2   ).reshape(__magic_name__			,		__magic_name__   )

     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.)
          lowercase :       Tuple       =			torch.load(os.path.join(__magic_name__			,		'''consolidated.00.pth'''   )			,		map_location='''cpu'''   )
     else:
          # Sharded
          lowercase :       str       =			[
              torch.load(os.path.join(__magic_name__			,		F"""consolidated.{i:02d}.pth"""   )			,		map_location='''cpu'''   )
              for i in range(__magic_name__   )
          ]
     lowercase :       Tuple       =			0
     lowercase :       str       =			{'''weight_map''': {}}
     for layer_i in range(__magic_name__   ):
          lowercase :       List[Any]       =			F"""pytorch_model-{layer_i + 1}-of-{n_layers + 1}.bin"""
          if model_size == "7B":
               # Unsharded
               lowercase :       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.

               lowercase :       List[Any]       =			{
                   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(),
               }
               lowercase :       Tuple       =			permute(
                   torch.cat(
                       [
                           loaded[i][F"""layers.{layer_i}.attention.wq.weight"""].view(__magic_name__			,		__magic_name__			,		__magic_name__   )
                           for i in range(__magic_name__   )
                       ]			,		dim=0			,		).reshape(__magic_name__			,		__magic_name__   )   )
               lowercase :       Any       =			permute(
                   torch.cat(
                       [
                           loaded[i][F"""layers.{layer_i}.attention.wk.weight"""].view(
                               __magic_name__			,		__magic_name__			,		__magic_name__   )
                           for i in range(__magic_name__   )
                       ]			,		dim=0			,		).reshape(__magic_name__			,		__magic_name__   )			,		__magic_name__			,		__magic_name__			,		__magic_name__			,		)
               lowercase :       Any       =			torch.cat(
                   [
                       loaded[i][F"""layers.{layer_i}.attention.wv.weight"""].view(
                           __magic_name__			,		__magic_name__			,		__magic_name__   )
                       for i in range(__magic_name__   )
                   ]			,		dim=0			,		).reshape(__magic_name__			,		__magic_name__   )

               lowercase :       int       =			torch.cat(
                   [loaded[i][F"""layers.{layer_i}.attention.wo.weight"""] for i in range(__magic_name__   )]			,		dim=1   )
               lowercase :       str       =			torch.cat(
                   [loaded[i][F"""layers.{layer_i}.feed_forward.w1.weight"""] for i in range(__magic_name__   )]			,		dim=0   )
               lowercase :       Any       =			torch.cat(
                   [loaded[i][F"""layers.{layer_i}.feed_forward.w2.weight"""] for i in range(__magic_name__   )]			,		dim=1   )
               lowercase :       List[Any]       =			torch.cat(
                   [loaded[i][F"""layers.{layer_i}.feed_forward.w3.weight"""] for i in range(__magic_name__   )]			,		dim=0   )

          lowercase :       List[Any]       =			inv_freq
          for k, v in state_dict.items():
               lowercase :       Tuple       =			filename
               param_count += v.numel()
          torch.save(__magic_name__			,		os.path.join(__magic_name__			,		__magic_name__   )   )

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

     for k, v in state_dict.items():
          lowercase :       Tuple       =			filename
          param_count += v.numel()
     torch.save(__magic_name__			,		os.path.join(__magic_name__			,		__magic_name__   )   )

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

     # 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.'''   )
     lowercase :       Dict       =			LlamaForCausalLM.from_pretrained(__magic_name__			,		torch_dtype=torch.floataa			,		low_cpu_mem_usage=__magic_name__   )
     # Avoid saving this as part of the config.
     del model.config._name_or_path

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






def  snake_case(    __magic_name__			,		__magic_name__   )			->							Optional[int]:





     '''simple docstring'''


     lowercase :       str       =			LlamaTokenizer if LlamaTokenizerFast is None else LlamaTokenizerFast
     print(F"""Saving a {tokenizer_class.__name__} to {tokenizer_path}."""   )
     lowercase :       Tuple       =			tokenizer_class(__magic_name__   )
     tokenizer.save_pretrained(__magic_name__   )






def  snake_case(    )			->							Optional[Any]:





     '''simple docstring'''


     lowercase :       List[str]       =			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=__magic_name__			,		help='''Whether or not to save using `safetensors`.'''   )
     lowercase :       List[str]       =			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			,		)
     lowercase :       Any       =			os.path.join(args.input_dir			,		'''tokenizer.model'''   )
     write_tokenizer(args.output_dir			,		__magic_name__   )


if __name__ == "__main__":
 main()