


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"
    )
    A   :					List[str]	=      None
A   :					Dict	=      {
    "7B": 11008,
    "13B": 13824,
    "30B": 17920,
    "65B": 22016,
    "70B": 28672,
}
A   :					Tuple	=      {
    "7B": 1,
    "7Bf": 1,
    "13B": 2,
    "13Bf": 2,
    "30B": 4,
    "65B": 8,
    "70B": 8,
    "70Bf": 8,
}

def    lowercase_	(	_A    :					Union[str, Any]					,      _A    :					Optional[int]=1					,      _A    :					List[Any]=256		):

      """simple docstring"""




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

def    lowercase_	(	_A    :					Union[str, Any]		):

      """simple docstring"""




      with open(_A					,      "r"		) as f:
            return json.load(_A		)

def    lowercase_	(	_A    :					Tuple					,      _A    :					Union[str, Any]		):

      """simple docstring"""




      with open(_A					,      "w"		) as f:
            json.dump(_A					,      _A		)

def    lowercase_	(	_A    :					Tuple					,      _A    :					int					,      _A    :					Dict					,      _A    :					List[str]=True		):

      """simple docstring"""




      os.makedirs(_A					,      exist_ok=_A		)
      lowerCamelCase__	:       Optional[Any]					=						os.path.join(_A					,      "tmp"		)
      os.makedirs(_A					,      exist_ok=_A		)

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

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

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

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

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

                  lowerCamelCase__	:       Union[str, Any]					=						torch.cat(
                      [loaded[i][F"layers.{layer_i}.attention.wo.weight"] for i in range(_A		)]					,      dim=1		)
                  lowerCamelCase__	:       Tuple					=						torch.cat(
                      [loaded[i][F"layers.{layer_i}.feed_forward.w1.weight"] for i in range(_A		)]					,      dim=0		)
                  lowerCamelCase__	:       Optional[Any]					=						torch.cat(
                      [loaded[i][F"layers.{layer_i}.feed_forward.w2.weight"] for i in range(_A		)]					,      dim=1		)
                  lowerCamelCase__	:       Any					=						torch.cat(
                      [loaded[i][F"layers.{layer_i}.feed_forward.w3.weight"] for i in range(_A		)]					,      dim=0		)

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

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

      for k, v in state_dict.items():
            lowerCamelCase__	:       Dict					=						filename
            param_count += v.numel()
      torch.save(_A					,      os.path.join(_A					,      _A		)		)

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

      # 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."		)
      lowerCamelCase__	:       List[Any]					=						LlamaForCausalLM.from_pretrained(_A					,      torch_dtype=torch.floataa					,      low_cpu_mem_usage=_A		)
      # Avoid saving this as part of the config.
      del model.config._name_or_path

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

def    lowercase_	(	_A    :					Any					,      _A    :					Optional[int]		):

      """simple docstring"""




      lowerCamelCase__	:       List[str]					=						LlamaTokenizer if LlamaTokenizerFast is None else LlamaTokenizerFast
      print(F"Saving a {tokenizer_class.__name__} to {tokenizer_path}."		)
      lowerCamelCase__	:       str					=						tokenizer_class(_A		)
      tokenizer.save_pretrained(_A		)

def    lowercase_	(	):

      """simple docstring"""




      lowerCamelCase__	:       Dict					=						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=_A					,      help="Whether or not to save using `safetensors`."		)
      lowerCamelCase__	:       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					,      )
      lowerCamelCase__	:       Optional[int]					=						os.path.join(args.input_dir					,      "tokenizer.model"		)
      write_tokenizer(args.output_dir					,      _A		)


if __name__ == "__main__":
    main()






def    lowercase_	(	_A    :					int					,      _A    :					int		):

      """simple docstring"""




      if a < 0 or b < 0:
            raise ValueError("the value of both inputs must be positive"		)

      lowerCamelCase__	:       List[str]					=						str(bin(_A		)		)[2:]  # remove the leading "0b"
      lowerCamelCase__	:       List[Any]					=						str(bin(_A		)		)[2:]  # remove the leading "0b"

      lowerCamelCase__	:       List[Any]					=						max(len(_A		)					,      len(_A		)		)

      return "0b" + "".join(
          str(int(char_a != char_b		)		)
          for char_a, char_b in zip(a_binary.zfill(_A		)					,      b_binary.zfill(_A		)		)		)


if __name__ == "__main__":
    import doctest

    doctest.testmod()


