
import functools
import logging
import os
import sys
import threading
from logging import (
    CRITICAL,  # NOQA
    DEBUG,  # NOQA
    ERROR,  # NOQA
    FATAL,  # NOQA
    INFO,  # NOQA
    NOTSET,  # NOQA
    WARN,  # NOQA
    WARNING,  # NOQA
)
from typing import Optional

import huggingface_hub.utils as hf_hub_utils
from tqdm import auto as tqdm_lib


lowerCAmelCase_						=			threading.Lock()
lowerCAmelCase_						=			None

lowerCAmelCase_						=			{
    'debug': logging.DEBUG,
    'info': logging.INFO,
    'warning': logging.WARNING,
    'error': logging.ERROR,
    'critical': logging.CRITICAL,
}

lowerCAmelCase_						=			logging.WARNING

lowerCAmelCase_						=			True






def  snake_case(    )			->							List[Any]:





     '''simple docstring'''


     lowercase :       Tuple       =			os.getenv('''TRANSFORMERS_VERBOSITY'''			,		__magic_name__   )
     if env_level_str:
          if env_level_str in log_levels:
               return log_levels[env_level_str]
          else:
               logging.getLogger().warning(
                   F"""Unknown option TRANSFORMERS_VERBOSITY={env_level_str}, """
                   F"""has to be one of: { ', '.join(log_levels.keys()   ) }"""   )
     return _default_log_level






def  snake_case(    )			->							str:





     '''simple docstring'''


     return __name__.split('''.'''   )[0]






def  snake_case(    )			->							logging.Logger:





     '''simple docstring'''


     return logging.getLogger(_get_library_name()   )






def  snake_case(    )			->							None:





     '''simple docstring'''


     global _default_handler

     with _lock:
          if _default_handler:
               # This library has already configured the library root logger.
               return
          lowercase :       List[Any]       =			logging.StreamHandler()  # Set sys.stderr as stream.
          lowercase :       Tuple       =			sys.stderr.flush

          # Apply our default configuration to the library root logger.
          lowercase :       str       =			_get_library_root_logger()
          library_root_logger.addHandler(_default_handler   )
          library_root_logger.setLevel(_get_default_logging_level()   )
          lowercase :       str       =			False






def  snake_case(    )			->							None:





     '''simple docstring'''


     global _default_handler

     with _lock:
          if not _default_handler:
               return

          lowercase :       Optional[Any]       =			_get_library_root_logger()
          library_root_logger.removeHandler(_default_handler   )
          library_root_logger.setLevel(logging.NOTSET   )
          lowercase :       str       =			None






def  snake_case(    )			->							int:





     '''simple docstring'''


     return log_levels






def  snake_case(    __magic_name__ = None   )			->							logging.Logger:





     '''simple docstring'''


     if name is None:
          lowercase :       Any       =			_get_library_name()

     _configure_library_root_logger()
     return logging.getLogger(__magic_name__   )






def  snake_case(    )			->							int:





     '''simple docstring'''


     _configure_library_root_logger()
     return _get_library_root_logger().getEffectiveLevel()






def  snake_case(    __magic_name__   )			->							None:





     '''simple docstring'''


     _configure_library_root_logger()
     _get_library_root_logger().setLevel(__magic_name__   )






def  snake_case(    )			->							str:





     '''simple docstring'''


     return set_verbosity(__magic_name__   )






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





     '''simple docstring'''


     return set_verbosity(__magic_name__   )






def  snake_case(    )			->							str:





     '''simple docstring'''


     return set_verbosity(__magic_name__   )






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





     '''simple docstring'''


     return set_verbosity(__magic_name__   )






def  snake_case(    )			->							None:





     '''simple docstring'''


     _configure_library_root_logger()

     assert _default_handler is not None
     _get_library_root_logger().removeHandler(_default_handler   )






def  snake_case(    )			->							None:





     '''simple docstring'''


     _configure_library_root_logger()

     assert _default_handler is not None
     _get_library_root_logger().addHandler(_default_handler   )






def  snake_case(    __magic_name__   )			->							None:





     '''simple docstring'''


     _configure_library_root_logger()

     assert handler is not None
     _get_library_root_logger().addHandler(__magic_name__   )






def  snake_case(    __magic_name__   )			->							None:





     '''simple docstring'''


     _configure_library_root_logger()

     assert handler is not None and handler not in _get_library_root_logger().handlers
     _get_library_root_logger().removeHandler(__magic_name__   )






def  snake_case(    )			->							None:





     '''simple docstring'''


     _configure_library_root_logger()
     lowercase :       List[Any]       =			False






def  snake_case(    )			->							None:





     '''simple docstring'''


     _configure_library_root_logger()
     lowercase :       Any       =			True






def  snake_case(    )			->							None:





     '''simple docstring'''


     lowercase :       Tuple       =			_get_library_root_logger().handlers

     for handler in handlers:
          lowercase :       Any       =			logging.Formatter('''[%(levelname)s|%(filename)s:%(lineno)s] %(asctime)s >> %(message)s'''   )
          handler.setFormatter(__magic_name__   )






def  snake_case(    )			->							None:





     '''simple docstring'''


     lowercase :       Dict       =			_get_library_root_logger().handlers

     for handler in handlers:
          handler.setFormatter(__magic_name__   )






def  snake_case(    self			,		*__magic_name__			,		**__magic_name__   )			->							str:





     '''simple docstring'''


     lowercase :       Tuple       =			os.getenv('''TRANSFORMERS_NO_ADVISORY_WARNINGS'''			,		__magic_name__   )
     if no_advisory_warnings:
          return
     self.warning(*__magic_name__			,		**__magic_name__   )


lowerCAmelCase_						=			warning_advice






@functools.lru_cache(__magic_name__   )
def  snake_case(    self			,		*__magic_name__			,		**__magic_name__   )			->							List[str]:





     '''simple docstring'''


     self.warning(*__magic_name__			,		**__magic_name__   )


lowerCAmelCase_						=			warning_once
class 						_A :






  def __init__(  self		:     Dict				,						*_A		:     Dict				,						**_A		:     Dict				)		->   List[str]:  # pylint: disable=unused-argument






       """simple docstring"""







       lowercase :       Tuple       =			args[0] if args else None






  def __iter__(  self		:     str				)		->   List[Any]:






       """simple docstring"""







       return iter(self._iterator				)






  def __getattr__(  self		:     Dict				,						_A		:     Dict				)		->   List[str]:






       """simple docstring"""







       def empty_fn(*_A		:     Any				,						**_A		:     Optional[int]				):  # pylint: disable=unused-argument
            return

       return empty_fn






  def __enter__(  self		:     Optional[Any]				)		->   Optional[int]:






       """simple docstring"""







       return self






  def __exit__(  self		:     int				,						_A		:     Union[str, Any]				,						_A		:     Tuple				,						_A		:     int				)		->   Any:






       """simple docstring"""







       return


class 						_A :






  def __call__(  self		:     List[Any]				,						*_A		:     Optional[Any]				,						**_A		:     List[Any]				)		->   List[Any]:






       """simple docstring"""







       if _tqdm_active:
            return tqdm_lib.tqdm(*_A				,						**_A				)
       else:
            return EmptyTqdm(*_A				,						**_A				)






  def  __a      (  self		:     Tuple				,						*_A		:     Tuple				,						**_A		:     List[str]				)		->   int:






       """simple docstring"""







       lowercase :       Optional[int]       =			None
       if _tqdm_active:
            return tqdm_lib.tqdm.set_lock(*_A				,						**_A				)






  def  __a      (  self		:     Tuple				)		->   str:






       """simple docstring"""







       if _tqdm_active:
            return tqdm_lib.tqdm.get_lock()


lowerCAmelCase_						=			_tqdm_cls()






def  snake_case(    )			->							bool:





     '''simple docstring'''


     global _tqdm_active
     return bool(_tqdm_active   )






def  snake_case(    )			->							Dict:





     '''simple docstring'''


     global _tqdm_active
     lowercase :       str       =			True
     hf_hub_utils.enable_progress_bars()






def  snake_case(    )			->							List[Any]:





     '''simple docstring'''


     global _tqdm_active
     lowercase :       List[str]       =			False
     hf_hub_utils.disable_progress_bars()





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()