

'''simple docstring'''

import argparse
import logging
import os
from pathlib import Path
from typing import Any, Dict

import pytorch_lightning as pl
from pytorch_lightning.utilities import rank_zero_info

from transformers import (
    AdamW,
    AutoConfig,
    AutoModel,
    AutoModelForPreTraining,
    AutoModelForQuestionAnswering,
    AutoModelForSeqaSeqLM,
    AutoModelForSequenceClassification,
    AutoModelForTokenClassification,
    AutoModelWithLMHead,
    AutoTokenizer,
    PretrainedConfig,
    PreTrainedTokenizer,
)
from transformers.optimization import (
    Adafactor,
    get_cosine_schedule_with_warmup,
    get_cosine_with_hard_restarts_schedule_with_warmup,
    get_linear_schedule_with_warmup,
    get_polynomial_decay_schedule_with_warmup,
)
from transformers.utils.versions import require_version


A_       						=							logging.getLogger(__name__)

require_version("pytorch_lightning>=1.0.4")

A_       						=							{
    "base": AutoModel,
    "sequence-classification": AutoModelForSequenceClassification,
    "question-answering": AutoModelForQuestionAnswering,
    "pretraining": AutoModelForPreTraining,
    "token-classification": AutoModelForTokenClassification,
    "language-modeling": AutoModelWithLMHead,
    "summarization": AutoModelForSeqaSeqLM,
    "translation": AutoModelForSeqaSeqLM,
}


# update this and the import above to support new schedulers from transformers.optimization
A_       						=							{
    "linear": get_linear_schedule_with_warmup,
    "cosine": get_cosine_schedule_with_warmup,
    "cosine_w_restarts": get_cosine_with_hard_restarts_schedule_with_warmup,
    "polynomial": get_polynomial_decay_schedule_with_warmup,
    # '': get_constant_schedule,             # not supported for now
    # '': get_constant_schedule_with_warmup, # not supported for now
}
A_       						=							sorted(arg_to_scheduler.keys())
A_       						=							"{" + ", ".join(arg_to_scheduler_choices) + "}"





class 							_snake_case (       pl.LightningModule						):
   def __init__( self			:      Any     ,SCREAMING_SNAKE_CASE__			:      argparse.Namespace     ,SCREAMING_SNAKE_CASE__			:      Optional[Any]=None     ,SCREAMING_SNAKE_CASE__			:      List[Any]="base"     ,SCREAMING_SNAKE_CASE__			:      Optional[Any]=None     ,SCREAMING_SNAKE_CASE__			:      Optional[Any]=None     ,SCREAMING_SNAKE_CASE__			:      Optional[Any]=None     ,**SCREAMING_SNAKE_CASE__			:      Union[str, Any]     ,):
     super().__init__()
     # TODO: move to self.save_hyperparameters()
     # self.save_hyperparameters()
     # can also expand arguments into trainer signature for easier reading

     self.save_hyperparameters(SCREAMING_SNAKE_CASE__							)
     SCREAMING_SNAKE_CASE:List[Any]						      =					0
     SCREAMING_SNAKE_CASE:int						      =					Path(self.hparams.output_dir							)
     SCREAMING_SNAKE_CASE:Optional[int]						      =					self.hparams.cache_dir if self.hparams.cache_dir else None
     if config is None:
       SCREAMING_SNAKE_CASE:List[Any]						      =					AutoConfig.from_pretrained(
           self.hparams.config_name if self.hparams.config_name else self.hparams.model_name_or_path     ,**({"num_labels": num_labels} if num_labels is not None else {})     ,cache_dir=SCREAMING_SNAKE_CASE__     ,**SCREAMING_SNAKE_CASE__     ,)
     else:
       SCREAMING_SNAKE_CASE:PretrainedConfig						      =					config

     SCREAMING_SNAKE_CASE:Tuple						      =					("encoder_layerdrop", "decoder_layerdrop", "dropout", "attention_dropout")
     for p in extra_model_params:
       if getattr(self.hparams     ,SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__							):
         assert hasattr(self.config     ,SCREAMING_SNAKE_CASE__							), F'''model config doesn\'t have a `{p}` attribute'''
         setattr(self.config     ,SCREAMING_SNAKE_CASE__     ,getattr(self.hparams     ,SCREAMING_SNAKE_CASE__							)							)

     if tokenizer is None:
       SCREAMING_SNAKE_CASE:str						      =					AutoTokenizer.from_pretrained(
           self.hparams.tokenizer_name if self.hparams.tokenizer_name else self.hparams.model_name_or_path     ,cache_dir=SCREAMING_SNAKE_CASE__     ,)
     else:
       SCREAMING_SNAKE_CASE:PreTrainedTokenizer						      =					tokenizer
     SCREAMING_SNAKE_CASE:Optional[int]						      =					MODEL_MODES[mode]
     if model is None:
       SCREAMING_SNAKE_CASE:Tuple						      =					self.model_type.from_pretrained(
           self.hparams.model_name_or_path     ,from_tf=bool(".ckpt" in self.hparams.model_name_or_path							)     ,config=self.config     ,cache_dir=SCREAMING_SNAKE_CASE__     ,)
     else:
       SCREAMING_SNAKE_CASE:Optional[Any]						      =					model
   def     __UpperCamelCase						( self			:      Dict     ,*SCREAMING_SNAKE_CASE__			:      Any     ,**SCREAMING_SNAKE_CASE__			:      str							):
     SCREAMING_SNAKE_CASE:Any						      =					self.model_type.from_pretrained(*SCREAMING_SNAKE_CASE__     ,**SCREAMING_SNAKE_CASE__							)
   def     __UpperCamelCase						( self			:      List[str]							):
     SCREAMING_SNAKE_CASE:int						      =					arg_to_scheduler[self.hparams.lr_scheduler]
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					get_schedule_func(
         self.opt     ,num_warmup_steps=self.hparams.warmup_steps     ,num_training_steps=self.total_steps()							)
     SCREAMING_SNAKE_CASE:str						      =					{"scheduler": scheduler, "interval": "step", "frequency": 1}
     return scheduler
   def     __UpperCamelCase						( self			:      str							):
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					self.model
     SCREAMING_SNAKE_CASE:Tuple						      =					["bias", "LayerNorm.weight"]
     SCREAMING_SNAKE_CASE:str						      =					[
         {
             "params": [
                 p for n, p in model.named_parameters() if not any(nd in n for nd in no_decay							)
             ],  # check this named paramters
             "weight_decay": self.hparams.weight_decay,
         },
         {
             "params": [p for n, p in model.named_parameters() if any(nd in n for nd in no_decay							)],
             "weight_decay": 0.0,
         },
     ]
     if self.hparams.adafactor:
       SCREAMING_SNAKE_CASE:Any						      =					Adafactor(
           SCREAMING_SNAKE_CASE__     ,lr=self.hparams.learning_rate     ,scale_parameter=SCREAMING_SNAKE_CASE__     ,relative_step=SCREAMING_SNAKE_CASE__							)

     else:
       SCREAMING_SNAKE_CASE:int						      =					AdamW(
           SCREAMING_SNAKE_CASE__     ,lr=self.hparams.learning_rate     ,eps=self.hparams.adam_epsilon							)
     SCREAMING_SNAKE_CASE:Any						      =					optimizer

     SCREAMING_SNAKE_CASE:Optional[Any]						      =					self.get_lr_scheduler()

     return [optimizer], [scheduler]
   def     __UpperCamelCase						( self			:      Union[str, Any]     ,SCREAMING_SNAKE_CASE__			:      Optional[Any]     ,SCREAMING_SNAKE_CASE__			:      List[str]							):
     return self.validation_step(SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__							)
   def     __UpperCamelCase						( self			:      Tuple     ,SCREAMING_SNAKE_CASE__			:      List[Any]							):
     return self.validation_end(SCREAMING_SNAKE_CASE__							)
   def     __UpperCamelCase						( self			:      List[str]							):
     SCREAMING_SNAKE_CASE:List[Any]						      =					max(1     ,self.hparams.gpus							)  # TODO: consider num_tpu_cores
     SCREAMING_SNAKE_CASE:Tuple						      =					self.hparams.train_batch_size * self.hparams.accumulate_grad_batches * num_devices
     return (self.dataset_size / effective_batch_size) * self.hparams.max_epochs
   def     __UpperCamelCase						( self			:      Optional[Any]     ,SCREAMING_SNAKE_CASE__			:      List[Any]							):
     if stage == "test":
       SCREAMING_SNAKE_CASE:List[Any]						      =					len(self.test_dataloader().dataset							)
     else:
       SCREAMING_SNAKE_CASE:Optional[Any]						      =					self.get_dataloader("train"     ,self.hparams.train_batch_size     ,shuffle=SCREAMING_SNAKE_CASE__							)
       SCREAMING_SNAKE_CASE:List[str]						      =					len(self.train_dataloader().dataset							)
   def     __UpperCamelCase						( self			:      Optional[Any]     ,SCREAMING_SNAKE_CASE__			:      str     ,SCREAMING_SNAKE_CASE__			:      int     ,SCREAMING_SNAKE_CASE__			:      bool = False							):
     raise NotImplementedError("You must implement this for your task"							)
   def     __UpperCamelCase						( self			:      Optional[int]							):
     return self.train_loader
   def     __UpperCamelCase						( self			:      List[Any]							):
     return self.get_dataloader("dev"     ,self.hparams.eval_batch_size     ,shuffle=SCREAMING_SNAKE_CASE__							)
   def     __UpperCamelCase						( self			:      List[str]							):
     return self.get_dataloader("test"     ,self.hparams.eval_batch_size     ,shuffle=SCREAMING_SNAKE_CASE__							)
   def     __UpperCamelCase						( self			:      Union[str, Any]     ,SCREAMING_SNAKE_CASE__			:      int							):
     return os.path.join(
         self.hparams.data_dir     ,"cached_{}_{}_{}".format(
             SCREAMING_SNAKE_CASE__     ,list(filter(SCREAMING_SNAKE_CASE__     ,self.hparams.model_name_or_path.split("/"							)							)							).pop()     ,str(self.hparams.max_seq_length							)     ,)     ,)
   @pl.utilities.rank_zero_only
   def     __UpperCamelCase						( self			:      Tuple     ,SCREAMING_SNAKE_CASE__			:      Dict[str, Any]							):
     SCREAMING_SNAKE_CASE:Dict						      =					self.output_dir.joinpath("best_tfmr"							)
     SCREAMING_SNAKE_CASE:int						      =					self.step_count
     self.model.save_pretrained(SCREAMING_SNAKE_CASE__							)
     self.tokenizer.save_pretrained(SCREAMING_SNAKE_CASE__							)







   @staticmethod
   def     __UpperCamelCase						( SCREAMING_SNAKE_CASE__			:      Optional[Any]     ,SCREAMING_SNAKE_CASE__			:      Dict							):
     parser.add_argument(
         "--model_name_or_path"     ,default=SCREAMING_SNAKE_CASE__     ,type=SCREAMING_SNAKE_CASE__     ,required=SCREAMING_SNAKE_CASE__     ,help="Path to pretrained model or model identifier from huggingface.co/models"     ,)
     parser.add_argument(
         "--config_name"     ,default=""     ,type=SCREAMING_SNAKE_CASE__     ,help="Pretrained config name or path if not the same as model_name"							)
     parser.add_argument(
         "--tokenizer_name"     ,default=SCREAMING_SNAKE_CASE__     ,type=SCREAMING_SNAKE_CASE__     ,help="Pretrained tokenizer name or path if not the same as model_name"     ,)
     parser.add_argument(
         "--cache_dir"     ,default=str(Path(SCREAMING_SNAKE_CASE__							).parent / "test_run" / "cache"							)     ,type=SCREAMING_SNAKE_CASE__     ,help="Where do you want to store the pre-trained models downloaded from huggingface.co"     ,)
     parser.add_argument(
         "--encoder_layerdrop"     ,type=SCREAMING_SNAKE_CASE__     ,help="Encoder layer dropout probability (Optional). Goes into model.config"     ,)
     parser.add_argument(
         "--decoder_layerdrop"     ,type=SCREAMING_SNAKE_CASE__     ,help="Decoder layer dropout probability (Optional). Goes into model.config"     ,)
     parser.add_argument(
         "--dropout"     ,type=SCREAMING_SNAKE_CASE__     ,help="Dropout probability (Optional). Goes into model.config"     ,)
     parser.add_argument(
         "--attention_dropout"     ,type=SCREAMING_SNAKE_CASE__     ,help="Attention dropout probability (Optional). Goes into model.config"     ,)
     parser.add_argument("--learning_rate"     ,default=5e-5     ,type=SCREAMING_SNAKE_CASE__     ,help="The initial learning rate for Adam."							)
     parser.add_argument(
         "--lr_scheduler"     ,default="linear"     ,choices=SCREAMING_SNAKE_CASE__     ,metavar=SCREAMING_SNAKE_CASE__     ,type=SCREAMING_SNAKE_CASE__     ,help="Learning rate scheduler"     ,)
     parser.add_argument("--weight_decay"     ,default=0.0     ,type=SCREAMING_SNAKE_CASE__     ,help="Weight decay if we apply some."							)
     parser.add_argument("--adam_epsilon"     ,default=1e-8     ,type=SCREAMING_SNAKE_CASE__     ,help="Epsilon for Adam optimizer."							)
     parser.add_argument("--warmup_steps"     ,default=0     ,type=SCREAMING_SNAKE_CASE__     ,help="Linear warmup over warmup_steps."							)
     parser.add_argument("--num_workers"     ,default=4     ,type=SCREAMING_SNAKE_CASE__     ,help="kwarg passed to DataLoader"							)
     parser.add_argument("--num_train_epochs"     ,dest="max_epochs"     ,default=3     ,type=SCREAMING_SNAKE_CASE__							)
     parser.add_argument("--train_batch_size"     ,default=32     ,type=SCREAMING_SNAKE_CASE__							)
     parser.add_argument("--eval_batch_size"     ,default=32     ,type=SCREAMING_SNAKE_CASE__							)
     parser.add_argument("--adafactor"     ,action="store_true"							)





class 							_snake_case (       pl.Callback						):
   def     __UpperCamelCase						( self			:      Tuple     ,SCREAMING_SNAKE_CASE__			:      List[str]     ,SCREAMING_SNAKE_CASE__			:      Any							):
     if (
         trainer.is_global_zero and trainer.global_rank == 0
     ):  # we initialize the retriever only on master worker with RAY. In new pytorch-lightning accelorators are removed.
       pl_module.model.rag.retriever.init_retrieval()  # better to use hook functions.





class 							_snake_case (       pl.Callback						):
   def     __UpperCamelCase						( self			:      Optional[Any]     ,SCREAMING_SNAKE_CASE__			:      str     ,SCREAMING_SNAKE_CASE__			:      Optional[int]							):
     # print(pl_module.model.rag)
     for name, param in pl_module.model.rag.named_parameters():
       if param.grad is None:
         print(SCREAMING_SNAKE_CASE__							)





class 							_snake_case (       pl.Callback						):
   def     __UpperCamelCase						( self			:      Optional[Any]     ,SCREAMING_SNAKE_CASE__			:      int     ,SCREAMING_SNAKE_CASE__			:      int							):
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					trainer.lr_schedulers[0]["scheduler"]
     SCREAMING_SNAKE_CASE:List[Any]						      =					{F'''lr_group_{i}''': lr for i, lr in enumerate(lr_scheduler.get_lr()							)}
     pl_module.logger.log_metrics(SCREAMING_SNAKE_CASE__							)
   def     __UpperCamelCase						( self			:      int     ,SCREAMING_SNAKE_CASE__			:      pl.Trainer     ,SCREAMING_SNAKE_CASE__			:      pl.LightningModule							):
     rank_zero_info("***** Validation results *****"							)
     SCREAMING_SNAKE_CASE:List[str]						      =					trainer.callback_metrics
     # Log results
     for key in sorted(SCREAMING_SNAKE_CASE__							):
       if key not in ["log", "progress_bar"]:
         rank_zero_info("{} = {}\n".format(SCREAMING_SNAKE_CASE__     ,str(metrics[key]							)							)							)







   def     __UpperCamelCase						( self			:      Optional[int]     ,SCREAMING_SNAKE_CASE__			:      pl.Trainer     ,SCREAMING_SNAKE_CASE__			:      pl.LightningModule							):
     rank_zero_info("***** Test results *****"							)
     SCREAMING_SNAKE_CASE:Any						      =					trainer.callback_metrics
     # Log and save results to file
     SCREAMING_SNAKE_CASE:Tuple						      =					os.path.join(pl_module.hparams.output_dir     ,"test_results.txt"							)
     with open(SCREAMING_SNAKE_CASE__     ,"w"							) as writer:
       for key in sorted(SCREAMING_SNAKE_CASE__							):
         if key not in ["log", "progress_bar"]:
           rank_zero_info("{} = {}\n".format(SCREAMING_SNAKE_CASE__     ,str(metrics[key]							)							)							)
           writer.write("{} = {}\n".format(SCREAMING_SNAKE_CASE__     ,str(metrics[key]							)							)							)





def    A_  ( snake_case      ,       snake_case ):
  #  To allow all pl args uncomment the following line
  #  parser = pl.Trainer.add_argparse_args(parser)
  parser.add_argument(
      "--output_dir"      ,       default=str(Path(snake_case ).parent / "test_run" / "model_checkpoints" )      ,       type=snake_case      ,       help="The output directory where the model predictions and checkpoints will be written."      ,       )
  parser.add_argument(
      "--fp16"      ,       action="store_true"      ,       help="Whether to use 16-bit (mixed) precision (through NVIDIA apex) instead of 32-bit"      ,       )

  parser.add_argument(
      "--fp16_opt_level"      ,       type=snake_case      ,       default="O2"      ,       help=(
          "For fp16: Apex AMP optimization level selected in ['O0', 'O1', 'O2', and 'O3']."
          "See details at https://nvidia.github.io/apex/amp.html"
      )      ,       )
  parser.add_argument("--n_tpu_cores"      ,       dest="tpu_cores"      ,       type=snake_case )
  parser.add_argument("--max_grad_norm"      ,       dest="gradient_clip_val"      ,       default=1.0      ,       type=snake_case      ,       help="Max gradient norm" )
  parser.add_argument("--do_train"      ,       action="store_true"      ,       help="Whether to run training." )
  parser.add_argument("--do_predict"      ,       action="store_true"      ,       help="Whether to run predictions on the test set." )
  parser.add_argument(
      "--gradient_accumulation_steps"      ,       dest="accumulate_grad_batches"      ,       type=snake_case      ,       default=1      ,       help="Number of updates steps to accumulate before performing a backward/update pass."      ,       )
  parser.add_argument("--seed"      ,       type=snake_case      ,       default=42      ,       help="random seed for initialization" )
  parser.add_argument(
      "--data_dir"      ,       default=str(Path(snake_case ).parent / "test_run" / "dummy-train-data" )      ,       type=snake_case      ,       help="The input data dir. Should contain the training files for the CoNLL-2003 NER task."      ,       )





def    A_  ( snake_case      ,       snake_case      ,       snake_case=None      ,       snake_case=True      ,       snake_case=[]      ,       snake_case=None      ,       snake_case=None      ,       **snake_case      ,       ):
  pl.seed_everything(args.seed )

  # init model
  SCREAMING_SNAKE_CASE:List[Any]						      =					Path(model.hparams.output_dir )
  odir.mkdir(exist_ok=snake_case )

  # add custom checkpoints
  if checkpoint_callback is None:
    SCREAMING_SNAKE_CASE:Union[str, Any]						      =					pl.callbacks.ModelCheckpoint(
        filepath=args.output_dir      ,       prefix="checkpoint"      ,       monitor="val_loss"      ,       mode="min"      ,       save_top_k=1 )
  if early_stopping_callback:
    extra_callbacks.append(snake_case )
  if logging_callback is None:
    SCREAMING_SNAKE_CASE:Optional[Any]						      =					LoggingCallback()

  SCREAMING_SNAKE_CASE:Optional[int]						      =					{}

  if args.fpaa:
    SCREAMING_SNAKE_CASE:List[str]						      =					16

  if args.gpus > 1:
    SCREAMING_SNAKE_CASE:str						      =					"auto"
    SCREAMING_SNAKE_CASE:Optional[int]						      =					"ddp"

  SCREAMING_SNAKE_CASE:Tuple						      =					args.accumulate_grad_batches
  SCREAMING_SNAKE_CASE:int						      =					None
  SCREAMING_SNAKE_CASE:str						      =					"auto"

  SCREAMING_SNAKE_CASE:int						      =					pl.Trainer.from_argparse_args(
      snake_case      ,       weights_summary=snake_case      ,       callbacks=[logging_callback] + extra_callbacks + [InitCallback()] + [checkpoint_callback]      ,       logger=snake_case      ,       val_check_interval=1      ,       num_sanity_val_steps=2      ,       **snake_case      ,       )

  if args.do_train:
    trainer.fit(snake_case )

  else:
    print("RAG modeling tests with new set functions successfuly executed!" )
  return trainer









'''simple docstring'''

from math import factorial





def    A_  ( snake_case      ,       snake_case ):

  # If either of the conditions are true, the function is being asked
  # to calculate a factorial of a negative number, which is not possible
  if n < k or k < 0:
    raise ValueError("Please enter positive integers for n and k where n >= k" )
  return factorial(snake_case ) // (factorial(snake_case ) * factorial(n - k ))


if __name__ == "__main__":
    print(
        "The number of five-card hands possible from a standard",
        f'''fifty-two card deck is: {combinations(52, 5)}\n''',
    )

    print(
        "If a class of 40 students must be arranged into groups of",
        f'''4 for group projects, there are {combinations(40, 4)} ways''',
        "to arrange them.\n",
    )

    print(
        "If 10 teams are competing in a Formula One race, there",
        f'''are {combinations(10, 3)} ways that first, second and''',
        "third place can be awarded.",
    )






