




'''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__		: Tuple												=  logging.getLogger(__name__)

require_version("""pytorch_lightning>=1.0.4""")

A__		: List[str]												=  {
    """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__		: Dict												=  {
    """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__		: Any												=  sorted(arg_to_scheduler.keys())
A__		: Any												=  """{""" + """, """.join(arg_to_scheduler_choices) + """}"""





class        UpperCAmelCase_  (pl.LightningModule  ):





    """simple docstring"""






    def __init__(							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_=None      ,					SCREAMING_SNAKE_CASE_="base"      ,					SCREAMING_SNAKE_CASE_=None      ,					SCREAMING_SNAKE_CASE_=None      ,					SCREAMING_SNAKE_CASE_=None      ,					**SCREAMING_SNAKE_CASE_      ,					)    ->					int:
         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_     )
         __lowerCamelCase       :     List[str]        =		0
         __lowerCamelCase       :     Union[str, Any]        =		Path(self.hparams.output_dir     )
         __lowerCamelCase       :     Union[str, Any]        =		self.hparams.cache_dir if self.hparams.cache_dir else None
         if config is None:
              __lowerCamelCase       :     Optional[int]        =		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:
              __lowerCamelCase       :     PretrainedConfig        =		config

         __lowerCamelCase       :     Dict        =		('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:
              __lowerCamelCase       :     Optional[int]        =		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:
              __lowerCamelCase       :     PreTrainedTokenizer        =		tokenizer
         __lowerCamelCase       :     int        =		MODEL_MODES[mode]
         if model is None:
              __lowerCamelCase       :     str        =		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:
              __lowerCamelCase       :     List[str]        =		model







    def 				lowercase_    (							self      ,					*SCREAMING_SNAKE_CASE_      ,					**SCREAMING_SNAKE_CASE_     )    ->					Optional[int]:
         __lowerCamelCase       :     Optional[int]        =		self.model_type.from_pretrained(*SCREAMING_SNAKE_CASE_      ,					**SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self     )    ->					Optional[Any]:
         __lowerCamelCase       :     Dict        =		arg_to_scheduler[self.hparams.lr_scheduler]
         __lowerCamelCase       :     Optional[int]        =		get_schedule_func(
             self.opt      ,					num_warmup_steps=self.hparams.warmup_steps      ,					num_training_steps=self.total_steps()     )
         __lowerCamelCase       :     str        =		{'scheduler': scheduler, 'interval': 'step', 'frequency': 1}
         return scheduler







    def 				lowercase_    (							self     )    ->					int:
         __lowerCamelCase       :     Dict        =		self.model
         __lowerCamelCase       :     Optional[Any]        =		['bias', 'LayerNorm.weight']
         __lowerCamelCase       :     str        =		[
             {
                 'params': [p for n, p in model.named_parameters() if not any(nd in n for nd in no_decay     )],
                 '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:
              __lowerCamelCase       :     Optional[int]        =		Adafactor(
                  SCREAMING_SNAKE_CASE_      ,					lr=self.hparams.learning_rate      ,					scale_parameter=SCREAMING_SNAKE_CASE_      ,					relative_step=SCREAMING_SNAKE_CASE_     )

         else:
              __lowerCamelCase       :     Dict        =		AdamW(
                  SCREAMING_SNAKE_CASE_      ,					lr=self.hparams.learning_rate      ,					eps=self.hparams.adam_epsilon     )
         __lowerCamelCase       :     Any        =		optimizer

         __lowerCamelCase       :     Dict        =		self.get_lr_scheduler()

         return [optimizer], [scheduler]







    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )    ->					Tuple:
         return self.validation_step(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_     )    ->					List[str]:
         return self.validation_end(SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self     )    ->					int:
         __lowerCamelCase       :     List[str]        =		max(1      ,					self.hparams.gpus     )  # TODO: consider num_tpu_cores
         __lowerCamelCase       :     Union[str, Any]        =		self.hparams.train_batch_size * self.hparams.accumulate_grad_batches * num_devices
         return (self.dataset_size / effective_batch_size) * self.hparams.max_epochs







    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_     )    ->					Optional[int]:
         if mode == "test":
              __lowerCamelCase       :     List[str]        =		len(self.test_dataloader().dataset     )
         else:
              __lowerCamelCase       :     Any        =		self.get_dataloader('train'      ,					self.hparams.train_batch_size      ,					shuffle=SCREAMING_SNAKE_CASE_     )
              __lowerCamelCase       :     List[Any]        =		len(self.train_dataloader().dataset     )







    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_ = False     )    ->					Dict:
         raise NotImplementedError('You must implement this for your task'     )







    def 				lowercase_    (							self     )    ->					List[Any]:
         return self.train_loader







    def 				lowercase_    (							self     )    ->					Dict:
         return self.get_dataloader('dev'      ,					self.hparams.eval_batch_size      ,					shuffle=SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self     )    ->					Any:
         return self.get_dataloader('test'      ,					self.hparams.eval_batch_size      ,					shuffle=SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_     )    ->					str:
         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 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_     )    ->					None:
         __lowerCamelCase       :     str        =		self.output_dir.joinpath('best_tfmr'     )
         __lowerCamelCase       :     Tuple        =		self.step_count
         self.model.save_pretrained(SCREAMING_SNAKE_CASE_     )
         self.tokenizer.save_pretrained(SCREAMING_SNAKE_CASE_     )







    @staticmethod
    def 				lowercase_    (							SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )    ->					Optional[Any]:
         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=''      ,					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        UpperCAmelCase_  (pl.Callback  ):





    """simple docstring"""






    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )    ->					Tuple:
         __lowerCamelCase       :     int        =		trainer.lr_schedulers[0]['scheduler']
         __lowerCamelCase       :     List[str]        =		{f'lr_group_{i}': lr for i, lr in enumerate(lr_scheduler.get_lr()     )}
         pl_module.logger.log_metrics(SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )    ->					Optional[int]:
         rank_zero_info('***** Validation results *****'     )
         __lowerCamelCase       :     int        =		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 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )    ->					int:
         rank_zero_info('***** Test results *****'     )
         __lowerCamelCase       :     List[str]        =		trainer.callback_metrics
         # Log and save results to file
         __lowerCamelCase       :     str        =		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       UpperCAmelCase__ (     UpperCAmelCase_   :    List[Any]				,   UpperCAmelCase_   :    str							)					->		None:
     #  To allow all pl args uncomment the following line
     #  parser = pl.Trainer.add_argparse_args(parser)
     parser.add_argument(
         '--output_dir'				,   default=UpperCAmelCase_				,   type=UpperCAmelCase_				,   required=UpperCAmelCase_				,   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=UpperCAmelCase_				,   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=UpperCAmelCase_							)
     parser.add_argument('--max_grad_norm'				,   dest='gradient_clip_val'				,   default=1.0				,   type=UpperCAmelCase_				,   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=UpperCAmelCase_				,   default=1				,   help='Number of updates steps to accumulate before performing a backward/update pass.'				,   )
     parser.add_argument('--seed'				,   type=UpperCAmelCase_				,   default=42				,   help='random seed for initialization'							)
     parser.add_argument(
         '--data_dir'				,   default=UpperCAmelCase_				,   type=UpperCAmelCase_				,   required=UpperCAmelCase_				,   help='The input data dir. Should contain the training files for the CoNLL-2003 NER task.'				,   )



def       UpperCAmelCase__ (     UpperCAmelCase_   :    BaseTransformer				,   UpperCAmelCase_   :    argparse.Namespace				,   UpperCAmelCase_   :    List[str]=None				,   UpperCAmelCase_   :    int=True				,   UpperCAmelCase_   :    str=[]				,   UpperCAmelCase_   :    Dict=None				,   UpperCAmelCase_   :    Optional[int]=None				,   **UpperCAmelCase_   :    Dict				,   )					->		Tuple:
     pl.seed_everything(args.seed							)

     # init model
     __lowerCamelCase       :     Dict        =		Path(model.hparams.output_dir							)
     odir.mkdir(exist_ok=UpperCAmelCase_							)

     # add custom checkpoints
     if checkpoint_callback is None:
          __lowerCamelCase       :     List[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(UpperCAmelCase_							)
     if logging_callback is None:
          __lowerCamelCase       :     List[str]        =		LoggingCallback()

     __lowerCamelCase       :     List[Any]        =		{}

     # TODO: remove with PyTorch 1.6 since pl uses native amp
     if args.fpaa:
          __lowerCamelCase       :     int        =		16
          __lowerCamelCase       :     Tuple        =		args.fpaa_opt_level

     if args.gpus > 1:
          __lowerCamelCase       :     List[Any]        =		'ddp'

     __lowerCamelCase       :     str        =		args.accumulate_grad_batches
     __lowerCamelCase       :     str        =		extra_train_kwargs.get('accelerator'				,   UpperCAmelCase_							)
     __lowerCamelCase       :     List[str]        =		extra_train_kwargs.get('profiler'				,   UpperCAmelCase_							)

     __lowerCamelCase       :     Optional[Any]        =		pl.Trainer.from_argparse_args(
         UpperCAmelCase_				,   weights_summary=UpperCAmelCase_				,   callbacks=[logging_callback] + extra_callbacks				,   logger=UpperCAmelCase_				,   checkpoint_callback=UpperCAmelCase_				,   **UpperCAmelCase_				,   )

     if args.do_train:
          trainer.fit(UpperCAmelCase_							)

     return trainer










'''simple docstring'''



from typing import Literal

A__		: Tuple												=  """ABCDEFGHIJKLMNOPQRSTUVWXYZ"""



def       UpperCAmelCase__ (     UpperCAmelCase_   :    str				,   UpperCAmelCase_   :    str				,   UpperCAmelCase_   :    Literal["encrypt", "decrypt"]							)					->		str:
     __lowerCamelCase       :     Dict        =		LETTERS if mode == 'decrypt' else key
     __lowerCamelCase       :     Tuple        =		key if mode == 'decrypt' else LETTERS
     __lowerCamelCase       :     Optional[int]        =		''
     # loop through each symbol in the message
     for symbol in message:
          if symbol.upper() in chars_a:
               # encrypt/decrypt the symbol
               __lowerCamelCase       :     Optional[int]        =		chars_a.find(symbol.upper()							)
               if symbol.isupper():
                    translated += chars_b[sym_index].upper()
               else:
                    translated += chars_b[sym_index].lower()
          else:
               # symbol is not in LETTERS, just add it
               translated += symbol
     return translated



def       UpperCAmelCase__ (     UpperCAmelCase_   :    str				,   UpperCAmelCase_   :    str							)					->		str:
     return translate_message(UpperCAmelCase_				,   UpperCAmelCase_				,   'encrypt'							)



def       UpperCAmelCase__ (     UpperCAmelCase_   :    str				,   UpperCAmelCase_   :    str							)					->		str:
     return translate_message(UpperCAmelCase_				,   UpperCAmelCase_				,   'decrypt'							)



def       UpperCAmelCase__ (     )					->		None:
     __lowerCamelCase       :     Dict        =		'Hello World'
     __lowerCamelCase       :     str        =		'QWERTYUIOPASDFGHJKLZXCVBNM'
     __lowerCamelCase       :     Any        =		'decrypt'  # set to 'encrypt' or 'decrypt'

     if mode == "encrypt":
          __lowerCamelCase       :     Optional[int]        =		encrypt_message(UpperCAmelCase_				,   UpperCAmelCase_							)
     elif mode == "decrypt":
          __lowerCamelCase       :     List[str]        =		decrypt_message(UpperCAmelCase_				,   UpperCAmelCase_							)
     print(F'Using the key {key}, the {mode}ed message is: {translated}'							)


if __name__ == "__main__":
      import doctest

      doctest.testmod()
      main()






