code: 
  
  
  """simple docstring"""
  import argparse
  import os
  from os.path import isfile
  
  import torch
  
  from transformers import GPTaConfig
  
  def 							_snake_case			(      __snake_case       :			List[str]	,   __snake_case       :			Optional[Any]	,   __snake_case       :			List[str]=0      ):
  
  
  
  
     """simple docstring"""
  
  
  
  
  
  
  
     if name is None:
        _lowerCamelCase  :				Optional[Any]      =		None
     else:
        _lowerCamelCase  :				List[Any]      =		""".""" * max(0	,   spaces - 2      ) + """# {:""" + str(50 - spaces      ) + """s}"""
        _lowerCamelCase  :				Tuple      =		fmt.format(__snake_case      )
  
     # Print and recurse (if needed).
     if isinstance(__snake_case	,   __snake_case      ):
        if msg is not None:
           print(__snake_case      )
        for k in val.keys():
           recursive_print(__snake_case	,   val[k]	,   spaces + 2      )
     elif isinstance(__snake_case	,   torch.Tensor      ):
        print(__snake_case	,   """:"""	,   val.size()      )
     else:
        print(__snake_case	,   """:"""	,   __snake_case      )
  
  def 							_snake_case			(      __snake_case       :			Tuple	,   __snake_case       :			Optional[Any]	,   __snake_case       :			int	,   __snake_case       :			int      ):
  
  
  
  
     """simple docstring"""
  
  
  
  
  
  
  
     _lowerCamelCase  :				Optional[int]      =		param.size()
     # other versions store [num_heads * num_splits * hidden_size, :]
     _lowerCamelCase  :				Union[str, Any]      =		(num_heads, num_splits, hidden_size) + input_shape[1:]
     _lowerCamelCase  :				Dict      =		param.view(*__snake_case      )
     _lowerCamelCase  :				Dict      =		param.transpose(0	,   1      ).contiguous()
     _lowerCamelCase  :				Optional[int]      =		param.view(*__snake_case      )
     return param
  
  def 							_snake_case			(      __snake_case       :			Tuple	,   __snake_case       :			Tuple      ):
  
  
  
  
     """simple docstring"""
  
  
  
  
  
  
  
     _lowerCamelCase  :				Any      =		config.n_positions
     _lowerCamelCase  :				Tuple      =		config.n_layer
     _lowerCamelCase  :				Tuple      =		config.vocab_size
     _lowerCamelCase  :				Any      =		config.n_head
     _lowerCamelCase  :				Dict      =		config.n_embd // config.n_head
  
     _lowerCamelCase  :				Tuple      =		sd_megatron["""model.language_model.embedding.word_embeddings.weight"""][:vocab_size, :]
     _lowerCamelCase  :				Optional[int]      =		{
         """transformer.wte.weight""": word_embeddings,
         """transformer.wpe.weight""": sd_megatron["""model.language_model.embedding.position_embeddings.weight"""],
         """transformer.ln_f.weight""": sd_megatron["""model.language_model.encoder.final_layernorm.weight"""],
         """transformer.ln_f.bias""": sd_megatron["""model.language_model.encoder.final_layernorm.bias"""],
     }
  
     _lowerCamelCase  :				Tuple      =		"""model.language_model.encoder.layers."""
     for i in range(__snake_case      ):
        _lowerCamelCase  :				Union[str, Any]      =		torch.tril(torch.ones((n_positions, n_positions)	,   dtype=torch.bool      )      )
        _lowerCamelCase  :				Optional[Any]      =		causal_mask.view(1	,   1	,   __snake_case	,   __snake_case      )
        _lowerCamelCase  :				Dict      =		causal_mask
        _lowerCamelCase  :				Tuple      =		torch.tensor(-1E4	,   dtype=torch.bfloataa      )
  
        _lowerCamelCase  :				Optional[int]      =		sd_megatron[F'{pf}{i}.input_layernorm.weight']
        _lowerCamelCase  :				Tuple      =		sd_megatron[F'{pf}{i}.input_layernorm.bias']
  
        _lowerCamelCase  :				List[str]      =		sd_megatron[F'{pf}{i}.self_attention.query_key_value.weight']
        _lowerCamelCase  :				List[Any]      =		fix_query_key_value_ordering(__snake_case	,   3	,   __snake_case	,   __snake_case      )
        _lowerCamelCase  :				List[Any]      =		vala.transpose(0	,   1      ).contiguous()
  
        _lowerCamelCase  :				Dict      =		sd_megatron[F'{pf}{i}.self_attention.query_key_value.bias']
        _lowerCamelCase  :				Dict      =		fix_query_key_value_ordering(__snake_case	,   3	,   __snake_case	,   __snake_case      )
        _lowerCamelCase  :				List[Any]      =		vala
  
        _lowerCamelCase  :				List[Any]      =		sd_megatron[F'{pf}{i}.self_attention.dense.weight'].transpose(
            0	,   1      )
        _lowerCamelCase  :				str      =		sd_megatron[F'{pf}{i}.self_attention.dense.bias']
        _lowerCamelCase  :				List[str]      =		sd_megatron[F'{pf}{i}.post_attention_layernorm.weight']
        _lowerCamelCase  :				Any      =		sd_megatron[F'{pf}{i}.post_attention_layernorm.bias']
        _lowerCamelCase  :				str      =		sd_megatron[F'{pf}{i}.mlp.dense_h_to_4h.weight'].transpose(0	,   1      )
        _lowerCamelCase  :				Tuple      =		sd_megatron[F'{pf}{i}.mlp.dense_h_to_4h.bias']
        _lowerCamelCase  :				List[Any]      =		sd_megatron[F'{pf}{i}.mlp.dense_4h_to_h.weight'].transpose(
            0	,   1      )
        _lowerCamelCase  :				List[str]      =		sd_megatron[F'{pf}{i}.mlp.dense_4h_to_h.bias']
  
     # For LM head, transformers' wants the matrix to weight embeddings.
     _lowerCamelCase  :				Any      =		word_embeddings
  
     return sd_hf
  
  def 							_snake_case			(      __snake_case       :			Optional[Any]	,   __snake_case       :			str      ):
  
  
  
  
     """simple docstring"""
  
  
  
  
  
  
  
     _lowerCamelCase  :				Union[str, Any]      =		64000
     _lowerCamelCase  :				Dict      =		config_megatron["""encoder_seq_length"""]
     _lowerCamelCase  :				Dict      =		config_megatron["""hidden_size"""]
     _lowerCamelCase  :				Union[str, Any]      =		config_megatron["""num_layers"""]
     _lowerCamelCase  :				str      =		config_megatron["""num_attention_heads"""]
     _lowerCamelCase  :				Union[str, Any]      =		config_megatron["""ffn_hidden_size"""]
     _lowerCamelCase  :				Optional[Any]      =		"""gelu"""
     _lowerCamelCase  :				int      =		0.1
     _lowerCamelCase  :				Optional[Any]      =		0.1
     _lowerCamelCase  :				Optional[int]      =		0.1
     _lowerCamelCase  :				Optional[int]      =		config_megatron["""layernorm_epsilon"""]  # 1e-5
     _lowerCamelCase  :				Dict      =		config_megatron["""init_method_std"""]  # 0.02
     _lowerCamelCase  :				Tuple      =		config_megatron["""apply_query_key_layer_scaling"""]  # True
     _lowerCamelCase  :				Union[str, Any]      =		True
     _lowerCamelCase  :				Optional[Any]      =		True
  
     # This identifies the 6.7B (7B) model which uses a different tokenizer
     if config_megatron["hidden_size"] == 4096:
        _lowerCamelCase  :				str      =		1  # <|endoftext|>
        _lowerCamelCase  :				Optional[Any]      =		1  # <|endoftext|>
        _lowerCamelCase  :				List[Any]      =		0  # <unk>
     else:
        _lowerCamelCase  :				int      =		2  # <s>
        _lowerCamelCase  :				Tuple      =		3  # <|endoftext|>
        _lowerCamelCase  :				Optional[int]      =		0  # <pad>
  
     return config_hf
  
  def 							_snake_case			(      __snake_case       :			str      ):
  
  
  
  
     """simple docstring"""
  
  
  
  
  
  
  
     print(__snake_case      )
  
     _lowerCamelCase  :				Any      =		args.checkpoint_path
     _lowerCamelCase  :				Tuple      =		args.save_path
     if isfile(__snake_case      ):
        raise FileNotFoundError(F'ERROR! could not find file {checkpoint_path}'      )
  
     # Load the model.
     _lowerCamelCase  :				Tuple      =		torch.load(__snake_case	,   map_location="""cpu"""      )
  
     # Load the config.
     _lowerCamelCase  :				Optional[Any]      =		checkpoint["""hyper_parameters"""]["""cfg"""]
     _lowerCamelCase  :				str      =		GPTaConfig()
     _lowerCamelCase  :				int      =		copy_config(config_hf=__snake_case	,   config_megatron=__snake_case      )
     _lowerCamelCase  :				Tuple      =		["""GPT2LMHeadModel"""]
  
     _lowerCamelCase  :				Union[str, Any]      =		checkpoint["""state_dict"""]
  
     # Convert.
     print("""Converting"""      )
     _lowerCamelCase  :				Any      =		convert_megatron_checkpoint(__snake_case	,   __snake_case      )
  
     # Print the structure of converted state dict.
     if args.print_checkpoint_structure:
        recursive_print(__snake_case	,   __snake_case      )
  
     _lowerCamelCase  :				Optional[Any]      =		"""GPTSw3Tokenizer"""
  
     # Store the config to file.
     print("""Saving config"""      )
     config_hf.save_pretrained(__snake_case      )
  
     # Store the state_dict to file.
     _lowerCamelCase  :				List[Any]      =		os.path.join(__snake_case	,   """pytorch_model.bin"""      )
     print(F'Saving checkpoint to "{output_checkpoint_file}"'      )
     torch.save(__snake_case	,   __snake_case      )
  
  
  if __name__ == "__main__":
   UpperCAmelCase   =	argparse.ArgumentParser()
   parser.add_argument(
       """--checkpoint_path""",
       type=str,
       required=True,
       help="""e.g. megatron_gpt--val_loss=2.42-step=38000-consumed_samples=54720000""",
   )
   parser.add_argument("""--save_path""", type=str, required=True, help="""e.g. /home/user/gpt-sw3/hf""")
   parser.add_argument("""--print-checkpoint-structure""", action="""store_true""")
   UpperCAmelCase   =	parser.parse_args()
   main(_args)
  
  
style_context: 
  
  
  """simple docstring"""
  import os
  
  from apache_beam.io.filesystems import FileSystems
  from apache_beam.pipeline import Pipeline
  
  from .logging import get_logger
  
  
  UpperCAmelCase   =	2 << 20  # 2mb
  UpperCAmelCase   =	get_logger(__name__)
  
  
  
  
  class   lowercase__							(				A_							):
  
  
  
  
  
     def 		UpperCamelCase_     (       self)     ->     Dict:
        _lowerCamelCase  :				Tuple      =		self._options.get_all_options().get("""runner""")
        return runner in [None, "DirectRunner", "PortableRunner"]
  
  
  
  
  
  
  
  def 							_snake_case			(      __snake_case       :			Union[str, Any]	,   __snake_case       :			List[Any]	,   __snake_case       :			Any=False      ):
  
  
  
  
     """simple docstring"""
  
  
  
  
  
  
  
     _lowerCamelCase  :				Any      =		FileSystems
     if fs.exists(__snake_case      ):
        if force_upload:
           logger.info(F'Remote path already exist: {remote_file_path}. Overwriting it as force_upload=True.'      )
        else:
           logger.info(F'Remote path already exist: {remote_file_path}. Skipping it as force_upload=False.'      )
           return
     with fs.create(__snake_case      ) as remote_file:
        with open(__snake_case	,   """rb"""      ) as local_file:
           _lowerCamelCase  :				Optional[Any]      =		local_file.read(__snake_case      )
           while chunk:
              remote_file.write(__snake_case      )
              _lowerCamelCase  :				Tuple      =		local_file.read(__snake_case      )
  
  
  
  
  
  def 							_snake_case			(      __snake_case       :			Union[str, Any]	,   __snake_case       :			List[Any]	,   __snake_case       :			Union[str, Any]=False      ):
  
  
  
  
     """simple docstring"""
  
  
  
  
  
  
  
     _lowerCamelCase  :				int      =		FileSystems
     if os.path.exists(__snake_case      ):
        if force_download:
           logger.info(F'Local path already exist: {remote_file_path}. Overwriting it as force_upload=True.'      )
        else:
           logger.info(F'Local path already exist: {remote_file_path}. Skipping it as force_upload=False.'      )
           return
     with fs.open(__snake_case      ) as remote_file:
        with open(__snake_case	,   """wb"""      ) as local_file:
           _lowerCamelCase  :				str      =		remote_file.read(__snake_case      )
           while chunk:
              local_file.write(__snake_case      )
              _lowerCamelCase  :				Any      =		remote_file.read(__snake_case      )
  
  