




'''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


lowercase						:   Union[str, Any]      						=							logging.getLogger(__name__)

require_version('''pytorch_lightning>=1.0.4''')

lowercase						:   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
lowercase						:   Tuple      						=							{
    '''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
}
lowercase						:   Union[str, Any]      						=							sorted(arg_to_scheduler.keys())
lowercase						:   List[str]      						=							'''{''' + ''', '''.join(arg_to_scheduler_choices) + '''}'''

class 	SCREAMING_SNAKE_CASE_		(							pl.LightningModule					):



 def __init__(	self ,		A ,		A=None ,		A="base" ,		A=None ,		A=None ,		A=None ,		**A ,		):






        '''simple docstring'''




        super().__init__()
        # TODO: move to self.save_hyperparameters()
        # self.save_hyperparameters()
        # can also expand arguments into trainer signature for easier reading

        self.save_hyperparameters(A			)
        lowercase	:		str    =   0
        lowercase	:		Optional[Any]    =   Path(self.hparams.output_dir			)
        lowercase	:		List[str]    =   self.hparams.cache_dir if self.hparams.cache_dir else None
        if config is None:
               lowercase	:		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=A ,		**A ,		)
        else:
               lowercase	:		PretrainedConfig    =   config

        lowercase	:		Optional[Any]    =   ('''encoder_layerdrop''', '''decoder_layerdrop''', '''dropout''', '''attention_dropout''')
        for p in extra_model_params:
               if getattr(self.hparams ,		A ,		A			):
                      assert hasattr(self.config ,		A			), f"""model config doesn't have a `{p}` attribute"""
                      setattr(self.config ,		A ,		getattr(self.hparams ,		A			)			)

        if tokenizer is None:
               lowercase	:		Dict    =   AutoTokenizer.from_pretrained(
                   self.hparams.tokenizer_name if self.hparams.tokenizer_name else self.hparams.model_name_or_path ,		cache_dir=A ,		)
        else:
               lowercase	:		PreTrainedTokenizer    =   tokenizer
        lowercase	:		Optional[Any]    =   MODEL_MODES[mode]
        if model is None:
               lowercase	:		Any    =   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=A ,		)
        else:
               lowercase	:		int    =   model



 def  SCREAMING_SNAKE_CASE(	self ,		*A ,		**A			):






        '''simple docstring'''




        lowercase	:		Tuple    =   self.model_type.from_pretrained(*A ,		**A			)



 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




        lowercase	:		List[Any]    =   arg_to_scheduler[self.hparams.lr_scheduler]
        lowercase	:		List[str]    =   get_schedule_func(
            self.opt ,		num_warmup_steps=self.hparams.warmup_steps ,		num_training_steps=self.total_steps()			)
        lowercase	:		Optional[int]    =   {'''scheduler''': scheduler, '''interval''': '''step''', '''frequency''': 1}
        return scheduler



 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




        lowercase	:		Optional[int]    =   self.model
        lowercase	:		List[Any]    =   ['''bias''', '''LayerNorm.weight''']
        lowercase	:		List[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:
               lowercase	:		Any    =   Adafactor(
                   A ,		lr=self.hparams.learning_rate ,		scale_parameter=A ,		relative_step=A			)

        else:
               lowercase	:		Optional[int]    =   AdamW(
                   A ,		lr=self.hparams.learning_rate ,		eps=self.hparams.adam_epsilon			)
        lowercase	:		List[str]    =   optimizer

        lowercase	:		Tuple    =   self.get_lr_scheduler()

        return [optimizer], [scheduler]



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A			):






        '''simple docstring'''




        return self.validation_step(A ,		A			)



 def  SCREAMING_SNAKE_CASE(	self ,		A			):






        '''simple docstring'''




        return self.validation_end(A			)



 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




        lowercase	:		Any    =   max(1 ,		self.hparams.gpus			)  # TODO: consider num_tpu_cores
        lowercase	:		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  SCREAMING_SNAKE_CASE(	self ,		A			):






        '''simple docstring'''




        if mode == "test":
               lowercase	:		str    =   len(self.test_dataloader().dataset			)
        else:
               lowercase	:		Union[str, Any]    =   self.get_dataloader('''train''' ,		self.hparams.train_batch_size ,		shuffle=A			)
               lowercase	:		Union[str, Any]    =   len(self.train_dataloader().dataset			)



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A = False			):






        '''simple docstring'''




        raise NotImplementedError('''You must implement this for your task'''			)



 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




        return self.train_loader



 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




        return self.get_dataloader('''dev''' ,		self.hparams.eval_batch_size ,		shuffle=A			)



 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




        return self.get_dataloader('''test''' ,		self.hparams.eval_batch_size ,		shuffle=A			)



 def  SCREAMING_SNAKE_CASE(	self ,		A			):






        '''simple docstring'''




        return os.path.join(
            self.hparams.data_dir ,		'''cached_{}_{}_{}'''.format(
                A ,		list(filter(A ,		self.hparams.model_name_or_path.split('''/'''			)			)			).pop() ,		str(self.hparams.max_seq_length			) ,		) ,		)



 @pl.utilities.rank_zero_only
 def  SCREAMING_SNAKE_CASE(	self ,		A			):






        '''simple docstring'''




        lowercase	:		Optional[Any]    =   self.output_dir.joinpath('''best_tfmr'''			)
        lowercase	:		Union[str, Any]    =   self.step_count
        self.model.save_pretrained(A			)
        self.tokenizer.save_pretrained(A			)





 @staticmethod
 def  SCREAMING_SNAKE_CASE(	A ,		A			):






        '''simple docstring'''




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






class 	SCREAMING_SNAKE_CASE_		(							pl.Callback					):



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		Any    =   trainer.lr_schedulers[0]['''scheduler''']
        lowercase	:		Tuple    =   {f"""lr_group_{i}""": lr for i, lr in enumerate(lr_scheduler.get_lr()			)}
        pl_module.logger.log_metrics(A			)



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A			):






        '''simple docstring'''




        rank_zero_info('''***** Validation results *****'''			)
        lowercase	:		Any    =   trainer.callback_metrics
        # Log results
        for key in sorted(A			):
               if key not in ["log", "progress_bar"]:
                      rank_zero_info('''{} = {}\n'''.format(A ,		str(metrics[key]			)			)			)





 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A			):






        '''simple docstring'''




        rank_zero_info('''***** Test results *****'''			)
        lowercase	:		Dict    =   trainer.callback_metrics
        # Log and save results to file
        lowercase	:		Tuple    =   os.path.join(pl_module.hparams.output_dir ,		'''test_results.txt'''			)
        with open(A ,		'''w'''			) as writer:
               for key in sorted(A			):
                      if key not in ["log", "progress_bar"]:
                             rank_zero_info('''{} = {}\n'''.format(A ,		str(metrics[key]			)			)			)
                             writer.write('''{} = {}\n'''.format(A ,		str(metrics[key]			)			)			)






def 		a_(			lowerCamelCase ,					lowerCamelCase   )					->None:






       '''simple docstring'''




       parser.add_argument(
           '''--output_dir''' ,					default=lowerCamelCase ,					type=lowerCamelCase ,					required=lowerCamelCase ,					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=lowerCamelCase ,					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=lowerCamelCase   )
       parser.add_argument('''--max_grad_norm''' ,					dest='''gradient_clip_val''' ,					default=1.0 ,					type=lowerCamelCase ,					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=lowerCamelCase ,					default=1 ,					help='''Number of updates steps to accumulate before performing a backward/update pass.''' ,					)
       parser.add_argument('''--seed''' ,					type=lowerCamelCase ,					default=42 ,					help='''random seed for initialization'''   )
       parser.add_argument(
           '''--data_dir''' ,					default=lowerCamelCase ,					type=lowerCamelCase ,					required=lowerCamelCase ,					help='''The input data dir. Should contain the training files for the CoNLL-2003 NER task.''' ,					)





def 		a_(			lowerCamelCase ,					lowerCamelCase ,					lowerCamelCase=None ,					lowerCamelCase=True ,					lowerCamelCase=[] ,					lowerCamelCase=None ,					lowerCamelCase=None ,					**lowerCamelCase ,					)					->Union[str, Any]:






       '''simple docstring'''




       pl.seed_everything(args.seed   )

       # init model
       lowercase	:		Optional[int]    =   Path(model.hparams.output_dir   )
       odir.mkdir(exist_ok=lowerCamelCase   )

       # add custom checkpoints
       if checkpoint_callback is None:
              lowercase	:		Dict    =   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(lowerCamelCase   )
       if logging_callback is None:
              lowercase	:		str    =   LoggingCallback()

       lowercase	:		List[str]    =   {}

       # TODO: remove with PyTorch 1.6 since pl uses native amp
       if args.fpaa:
              lowercase	:		List[Any]    =   16
              lowercase	:		Tuple    =   args.fpaa_opt_level

       if args.gpus > 1:
              lowercase	:		Optional[int]    =   '''ddp'''

       lowercase	:		int    =   args.accumulate_grad_batches
       lowercase	:		List[Any]    =   extra_train_kwargs.get('''accelerator''' ,					lowerCamelCase   )
       lowercase	:		Union[str, Any]    =   extra_train_kwargs.get('''profiler''' ,					lowerCamelCase   )

       lowercase	:		List[str]    =   pl.Trainer.from_argparse_args(
           lowerCamelCase ,					weights_summary=lowerCamelCase ,					callbacks=[logging_callback] + extra_callbacks ,					logger=lowerCamelCase ,					checkpoint_callback=lowerCamelCase ,					**lowerCamelCase ,					)

       if args.do_train:
              trainer.fit(lowerCamelCase   )

       return trainer










'''simple docstring'''






import timeit
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_auto import MODEL_MAPPING, MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_torch_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_torch_available():
      import torch

      from .benchmark_args import PyTorchBenchmarkArguments


if is_pyanvml_available():
      import pyanvml.pyanvml as nvml


lowercase						:   List[Any]      						=							logging.get_logger(__name__)

class 	SCREAMING_SNAKE_CASE_		(							lowerCamelCase_					):
 __snake_case						:  PyTorchBenchmarkArguments
 __snake_case						:  PretrainedConfig
 __snake_case						:  str       =   "PyTorch"



 @property
 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




        return torch.__version__



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		str    =   self._prepare_inference_func(A ,		A ,		A			)
        return self._measure_speed(_inference			)



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		int    =   self._prepare_inference_func(A ,		A ,		A			)
        return self._measure_memory(_inference			)



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		Union[str, Any]    =   self._prepare_train_func(A ,		A ,		A			)
        return self._measure_speed(_train			)



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		Any    =   self._prepare_train_func(A ,		A ,		A			)
        return self._measure_memory(_train			)



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		Any    =   self.config_dict[model_name]

        if self.args.torchscript:
               lowercase	:		Any    =   True

        lowercase	:		Union[str, Any]    =   (
            hasattr(A ,		'''architectures'''			)
            and isinstance(config.architectures ,		A			)
            and len(config.architectures			) > 0
        )
        if not self.args.only_pretrain_model and has_model_class_in_config:
               try:
                      lowercase	:		Any    =   config.architectures[0]
                      lowercase	:		Dict    =   __import__('''transformers''' ,		fromlist=[model_class]			)
                      lowercase	:		List[str]    =   getattr(A ,		A			)
                      lowercase	:		Dict    =   model_cls(A			)
               except ImportError:
                      raise ImportError(
                          f"""{model_class} does not exist. If you just want to test the pretrained model, you might want to"""
                          ''' set `--only_pretrain_model` or `args.only_pretrain_model=True`.'''			)
        else:
               lowercase	:		Optional[int]    =   MODEL_MAPPING[config.__class__](A			)

        model.eval()
        model.to(self.args.device			)

        # encoder-decoder has vocab size saved differently
        lowercase	:		str    =   config.vocab_size if hasattr(A ,		'''vocab_size'''			) else config.encoder.vocab_size
        lowercase	:		Union[str, Any]    =   torch.randint(A ,		(batch_size, sequence_length) ,		dtype=torch.long ,		device=self.args.device			)

        if self.args.fpaa:
               logger.info('''Running training in Mixed Precision...'''			)
               if not self.args.is_gpu:
                      raise ValueError('''Mixed precision is possible only for GPU.'''			)
               # amp seems to have memory leaks so that memory usage
               # is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
               model.half()

        if self.args.torchscript:
               with torch.no_grad():
                      lowercase	:		int    =   torch.jit.trace(A ,		A			)
        else:
               lowercase	:		Any    =   model

        def encoder_decoder_forward():
               with torch.no_grad():
                      lowercase	:		int    =   inference_model(A ,		decoder_input_ids=A			)
               return outputs

        def encoder_forward():
               with torch.no_grad():
                      lowercase	:		Tuple    =   inference_model(A			)
               return outputs

        lowercase	:		int    =   encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
        return _forward



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		Optional[int]    =   self.config_dict[model_name]

        lowercase	:		Optional[Any]    =   (
            hasattr(A ,		'''architectures'''			)
            and isinstance(config.architectures ,		A			)
            and len(config.architectures			) > 0
        )
        if not self.args.only_pretrain_model and has_model_class_in_config:
               try:
                      lowercase	:		str    =   config.architectures[0]
                      lowercase	:		List[str]    =   __import__('''transformers''' ,		fromlist=[model_class]			)
                      lowercase	:		int    =   getattr(A ,		A			)
                      lowercase	:		Any    =   model_cls(A			)
               except ImportError:
                      raise ImportError(
                          f"""{model_class} does not exist. If you just want to test the pretrained model, you might want to"""
                          ''' set `--only_pretrain_model` or `args.only_pretrain_model=True`.'''			)
        else:
               lowercase	:		Union[str, Any]    =   MODEL_WITH_LM_HEAD_MAPPING[config.__class__](A			)

        if self.args.torchscript:
               raise NotImplementedError('''Training for torchscript is currently not implemented'''			)
        else:
               lowercase	:		Dict    =   model

        model.train()
        model.to(self.args.device			)

        # encoder-decoder has vocab size saved differently
        lowercase	:		Union[str, Any]    =   config.vocab_size if hasattr(A ,		'''vocab_size'''			) else config.encoder.vocab_size
        lowercase	:		List[Any]    =   torch.randint(A ,		(batch_size, sequence_length) ,		dtype=torch.long ,		device=self.args.device			)

        if self.args.fpaa:
               logger.info('''Running training in Mixed Precision...'''			)
               if not self.args.is_gpu:
                      raise ValueError('''Mixed precision is possible only for GPU.'''			)

               # amp seems to have memory leaks so that memory usage
               # is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
               model.half()

        def compute_loss_and_backprob_encoder():
               lowercase	:		Optional[Any]    =   train_model(A ,		labels=A			)[0]
               loss.backward()
               return loss

        def compute_loss_and_backprob_encoder_decoder():
               lowercase	:		List[Any]    =   train_model(A ,		decoder_input_ids=A ,		labels=A			)[0]
               loss.backward()
               return loss

        lowercase	:		Any    =   (
            compute_loss_and_backprob_encoder_decoder
            if config.is_encoder_decoder
            else compute_loss_and_backprob_encoder
        )
        return _train



 def  SCREAMING_SNAKE_CASE(	self ,		A			):






        '''simple docstring'''




        try:
               if self.args.is_tpu or self.args.torchscript:
                      # run additional 10 times to stabilize compilation for tpu and torchscript
                      logger.info('''Do inference on TPU or torchscript. Running model 5 times to stabilize compilation'''			)
                      timeit.repeat(
                          A ,		repeat=1 ,		number=5 ,		)

               # as written in https://docs.python.org/2/library/timeit.html#timeit.Timer.repeat, min should be taken rather than the average
               lowercase	:		str    =   timeit.repeat(
                   A ,		repeat=self.args.repeat ,		number=10 ,		)

               if self.args.is_tpu and self.args.torch_xla_tpu_print_metrics:
                      import torch_xla.debug.metrics as met

                      self.print_fn(met.metrics_report()			)

               return min(A			) / 1_0.0
        except RuntimeError as e:
               self.print_fn(f"""Doesn't fit on GPU. {e}"""			)
               return "N/A"





 def  SCREAMING_SNAKE_CASE(	self ,		A			):






        '''simple docstring'''




        try:
               if self.args.trace_memory_line_by_line:
                      lowercase	:		int    =   start_memory_tracing('''transformers'''			)

               if self.args.is_tpu:
                      # tpu
                      raise NotImplementedError(
                          '''Memory Benchmarking is currently not implemented for TPU. Please disable memory benchmarking with'''
                          ''' `--no-memory` or `args.memory=False`'''			)
               elif self.args.is_gpu:
                      if not is_pyanvml_available():
                             logger.warning(
                                 '''py3nvml not installed, we won\'t log GPU memory usage. '''
                                 '''Install py3nvml (pip install py3nvml) to log information about GPU.'''			)
                             lowercase	:		str    =   '''N/A'''
                      else:
                             logger.info(
                                 '''Measuring total GPU usage on GPU device. Make sure to not have additional processes running'''
                                 ''' on the same GPU.'''			)
                             # init nvml
                             nvml.nvmlInit()
                             func()
                             lowercase	:		List[Any]    =   nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx			)
                             lowercase	:		List[Any]    =   nvml.nvmlDeviceGetMemoryInfo(A			)
                             lowercase	:		int    =   meminfo.used
                             lowercase	:		Dict    =   Memory(A			)
                             # shutdown nvml
                             nvml.nvmlShutdown()
               else:
                      # cpu
                      lowercase	:		Optional[Any]    =   measure_peak_memory_cpu(A			)
                      lowercase	:		Union[str, Any]    =   Memory(A			) if isinstance(A ,		A			) else memory_bytes

               if self.args.trace_memory_line_by_line:
                      lowercase	:		Tuple    =   stop_memory_tracing(A			)
               else:
                      lowercase	:		int    =   None

               return memory, summary
        except RuntimeError as e:
               self.print_fn(f"""Doesn't fit on GPU. {e}"""			)
               return "N/A", None




