
"""simple docstring"""




import math
from enum import Enum
from typing import Optional, Union

from torch.optim import Optimizer
from torch.optim.lr_scheduler import LambdaLR

from .utils import logging


A__      :	int          =    logging.get_logger(__name__)







class   _lowercase						(       lowerCAmelCase_				):


     '''simple docstring'''





     _A								=							'linear'
     _A								=							'cosine'
     _A								=							'cosine_with_restarts'
     _A								=							'polynomial'
     _A								=							'constant'
     _A								=							'constant_with_warmup'
     _A								=							'piecewise_constant'





def 							a__ (				lowerCAmelCase	:       Optimizer       ,    lowerCAmelCase	:       int = -1		):






     '''simple docstring'''
     return LambdaLR(lowerCAmelCase       ,    lambda lowerCAmelCase		: 1       ,    last_epoch=lowerCAmelCase		)



def 							a__ (				lowerCAmelCase	:       Optimizer       ,    lowerCAmelCase	:       int       ,    lowerCAmelCase	:       int = -1		):






     '''simple docstring'''

     def lr_lambda(lowerCAmelCase	:       int		):
          if current_step < num_warmup_steps:
               return float(lowerCAmelCase		) / float(max(1.0       ,    lowerCAmelCase		)		)
          return 1.0

     return LambdaLR(lowerCAmelCase       ,    lowerCAmelCase       ,    last_epoch=lowerCAmelCase		)



def 							a__ (				lowerCAmelCase	:       Optimizer       ,    lowerCAmelCase	:       str       ,    lowerCAmelCase	:       int = -1		):






     '''simple docstring'''
     UpperCAmelCase__							: Union[str, Any] 							=						{}
     UpperCAmelCase__							: Tuple 							=						step_rules.split(","		)
     for rule_str in rule_list[:-1]:
          UpperCAmelCase__							,	UpperCAmelCase__							: Dict 							=						rule_str.split(":"		)
          UpperCAmelCase__							: Any 							=						int(lowerCAmelCase		)
          UpperCAmelCase__							: Dict 							=						float(lowerCAmelCase		)
          UpperCAmelCase__							: Union[str, Any] 							=						value
     UpperCAmelCase__							: Optional[int] 							=						float(rule_list[-1]		)

     def create_rules_function(lowerCAmelCase	:       List[Any]       ,    lowerCAmelCase	:       Tuple		):
          def rule_func(lowerCAmelCase	:       int		) -> float:
               UpperCAmelCase__							: Any 							=						sorted(rules_dict.keys()		)
               for i, sorted_step in enumerate(lowerCAmelCase		):
                    if steps < sorted_step:
                         return rules_dict[sorted_steps[i]]
               return last_lr_multiple

          return rule_func

     UpperCAmelCase__							: int 							=						create_rules_function(lowerCAmelCase       ,    lowerCAmelCase		)

     return LambdaLR(lowerCAmelCase       ,    lowerCAmelCase       ,    last_epoch=lowerCAmelCase		)



def 							a__ (				lowerCAmelCase	:       Any       ,    lowerCAmelCase	:       str       ,    lowerCAmelCase	:       str       ,    lowerCAmelCase	:       str=-1		):






     '''simple docstring'''

     def lr_lambda(lowerCAmelCase	:       int		):
          if current_step < num_warmup_steps:
               return float(lowerCAmelCase		) / float(max(1       ,    lowerCAmelCase		)		)
          return max(
              0.0       ,    float(num_training_steps - current_step		) / float(max(1       ,    num_training_steps - num_warmup_steps		)		)		)

     return LambdaLR(lowerCAmelCase       ,    lowerCAmelCase       ,    lowerCAmelCase		)



def 							a__ (				lowerCAmelCase	:       Optimizer       ,    lowerCAmelCase	:       int       ,    lowerCAmelCase	:       int       ,    lowerCAmelCase	:       float = 0.5       ,    lowerCAmelCase	:       int = -1		):






     '''simple docstring'''

     def lr_lambda(lowerCAmelCase	:       int		):
          if current_step < num_warmup_steps:
               return float(lowerCAmelCase		) / float(max(1       ,    lowerCAmelCase		)		)
          UpperCAmelCase__							: List[str] 							=						float(current_step - num_warmup_steps		) / float(max(1       ,    num_training_steps - num_warmup_steps		)		)
          return max(0.0       ,    0.5 * (1.0 + math.cos(math.pi * float(lowerCAmelCase		) * 2.0 * progress		))		)

     return LambdaLR(lowerCAmelCase       ,    lowerCAmelCase       ,    lowerCAmelCase		)



def 							a__ (				lowerCAmelCase	:       Optimizer       ,    lowerCAmelCase	:       int       ,    lowerCAmelCase	:       int       ,    lowerCAmelCase	:       int = 1       ,    lowerCAmelCase	:       int = -1		):






     '''simple docstring'''

     def lr_lambda(lowerCAmelCase	:       Tuple		):
          if current_step < num_warmup_steps:
               return float(lowerCAmelCase		) / float(max(1       ,    lowerCAmelCase		)		)
          UpperCAmelCase__							: int 							=						float(current_step - num_warmup_steps		) / float(max(1       ,    num_training_steps - num_warmup_steps		)		)
          if progress >= 1.0:
               return 0.0
          return max(0.0       ,    0.5 * (1.0 + math.cos(math.pi * ((float(lowerCAmelCase		) * progress) % 1.0)		))		)

     return LambdaLR(lowerCAmelCase       ,    lowerCAmelCase       ,    lowerCAmelCase		)



def 							a__ (				lowerCAmelCase	:       Optional[int]       ,    lowerCAmelCase	:       List[Any]       ,    lowerCAmelCase	:       List[str]       ,    lowerCAmelCase	:       Any=1E-7       ,    lowerCAmelCase	:       Union[str, Any]=1.0       ,    lowerCAmelCase	:       List[Any]=-1		):






     '''simple docstring'''
     UpperCAmelCase__							: Union[str, Any] 							=						optimizer.defaults["lr"]
     if not (lr_init > lr_end):
          raise ValueError(F"lr_end ({lr_end}) must be be smaller than initial lr ({lr_init})"		)

     def lr_lambda(lowerCAmelCase	:       int		):
          if current_step < num_warmup_steps:
               return float(lowerCAmelCase		) / float(max(1       ,    lowerCAmelCase		)		)
          elif current_step > num_training_steps:
               return lr_end / lr_init  # as LambdaLR multiplies by lr_init
          else:
               UpperCAmelCase__							: Tuple 							=						lr_init - lr_end
               UpperCAmelCase__							: Optional[int] 							=						num_training_steps - num_warmup_steps
               UpperCAmelCase__							: Optional[Any] 							=						1 - (current_step - num_warmup_steps) / decay_steps
               UpperCAmelCase__							: Optional[int] 							=						lr_range * pct_remaining**power + lr_end
               return decay / lr_init  # as LambdaLR multiplies by lr_init

     return LambdaLR(lowerCAmelCase       ,    lowerCAmelCase       ,    lowerCAmelCase		)


A__      :	str          =    {
    SchedulerType.LINEAR: get_linear_schedule_with_warmup,
    SchedulerType.COSINE: get_cosine_schedule_with_warmup,
    SchedulerType.COSINE_WITH_RESTARTS: get_cosine_with_hard_restarts_schedule_with_warmup,
    SchedulerType.POLYNOMIAL: get_polynomial_decay_schedule_with_warmup,
    SchedulerType.CONSTANT: get_constant_schedule,
    SchedulerType.CONSTANT_WITH_WARMUP: get_constant_schedule_with_warmup,
    SchedulerType.PIECEWISE_CONSTANT: get_piecewise_constant_schedule,
}



def 							a__ (				lowerCAmelCase	:       Union[str, SchedulerType]       ,    lowerCAmelCase	:       Optimizer       ,    lowerCAmelCase	:       Optional[str] = None       ,    lowerCAmelCase	:       Optional[int] = None       ,    lowerCAmelCase	:       Optional[int] = None       ,    lowerCAmelCase	:       int = 1       ,    lowerCAmelCase	:       float = 1.0       ,    lowerCAmelCase	:       int = -1       ,    ):






     '''simple docstring'''
     UpperCAmelCase__							: Optional[Any] 							=						SchedulerType(lowerCAmelCase		)
     UpperCAmelCase__							: Tuple 							=						TYPE_TO_SCHEDULER_FUNCTION[name]
     if name == SchedulerType.CONSTANT:
          return schedule_func(lowerCAmelCase       ,    last_epoch=lowerCAmelCase		)

     if name == SchedulerType.PIECEWISE_CONSTANT:
          return schedule_func(lowerCAmelCase       ,    step_rules=lowerCAmelCase       ,    last_epoch=lowerCAmelCase		)

     # All other schedulers require `num_warmup_steps`
     if num_warmup_steps is None:
          raise ValueError(F"{name} requires `num_warmup_steps`, please provide that argument."		)

     if name == SchedulerType.CONSTANT_WITH_WARMUP:
          return schedule_func(lowerCAmelCase       ,    num_warmup_steps=lowerCAmelCase       ,    last_epoch=lowerCAmelCase		)

     # All other schedulers require `num_training_steps`
     if num_training_steps is None:
          raise ValueError(F"{name} requires `num_training_steps`, please provide that argument."		)

     if name == SchedulerType.COSINE_WITH_RESTARTS:
          return schedule_func(
              lowerCAmelCase       ,    num_warmup_steps=lowerCAmelCase       ,    num_training_steps=lowerCAmelCase       ,    num_cycles=lowerCAmelCase       ,    last_epoch=lowerCAmelCase       ,    )

     if name == SchedulerType.POLYNOMIAL:
          return schedule_func(
              lowerCAmelCase       ,    num_warmup_steps=lowerCAmelCase       ,    num_training_steps=lowerCAmelCase       ,    power=lowerCAmelCase       ,    last_epoch=lowerCAmelCase       ,    )

     return schedule_func(
         lowerCAmelCase       ,    num_warmup_steps=lowerCAmelCase       ,    num_training_steps=lowerCAmelCase       ,    last_epoch=lowerCAmelCase		)





"""simple docstring"""




import argparse

import torch

from transformers import GPTaConfig, GPTaModel, load_tf_weights_in_gpta
from transformers.utils import CONFIG_NAME, WEIGHTS_NAME, logging


logging.set_verbosity_info()



def 							a__ (				lowerCAmelCase	:       List[Any]       ,    lowerCAmelCase	:       Any       ,    lowerCAmelCase	:       List[Any]		):






     '''simple docstring'''
     # Construct model
     if gpta_config_file == "":
          UpperCAmelCase__							: Optional[int] 							=						GPTaConfig()
     else:
          UpperCAmelCase__							: Dict 							=						GPTaConfig.from_json_file(lowerCAmelCase		)
     UpperCAmelCase__							: Optional[Any] 							=						GPTaModel(lowerCAmelCase		)

     # Load weights from numpy
     load_tf_weights_in_gpta(lowerCAmelCase       ,    lowerCAmelCase       ,    lowerCAmelCase		)

     # Save pytorch-model
     UpperCAmelCase__							: Optional[int] 							=						pytorch_dump_folder_path + "/" + WEIGHTS_NAME
     UpperCAmelCase__							: Any 							=						pytorch_dump_folder_path + "/" + CONFIG_NAME
     print(F"Save PyTorch model to {pytorch_weights_dump_path}"		)
     torch.save(model.state_dict()       ,    lowerCAmelCase		)
     print(F"Save configuration file to {pytorch_config_dump_path}"		)
     with open(lowerCAmelCase       ,    "w"       ,    encoding="utf-8"		) as f:
          f.write(config.to_json_string()		)


if __name__ == "__main__":
    A__      :	Optional[int]          =    argparse.ArgumentParser()
    # Required parameters
    parser.add_argument(
        """--gpt2_checkpoint_path""", default=None, type=str, required=True, help="""Path to the TensorFlow checkpoint path."""
    )
    parser.add_argument(
        """--pytorch_dump_folder_path""", default=None, type=str, required=True, help="""Path to the output PyTorch model."""
    )
    parser.add_argument(
        """--gpt2_config_file""",
        default="""""",
        type=str,
        help=(
            """An optional config json file corresponding to the pre-trained OpenAI model. \n"""
            """This specifies the model architecture."""
        ),
    )
    A__      :	Optional[Any]          =    parser.parse_args()
    convert_gpta_checkpoint_to_pytorch(args.gpta_checkpoint_path, args.gpta_config_file, args.pytorch_dump_folder_path)



