

from typing import Any, Dict, List, Optional, Tuple, Union

import torch
from torch import nn
from torch.utils.data import DistributedSampler, RandomSampler

from transformers import PreTrainedModel, Trainer, logging
from transformers.integrations import is_fairscale_available
from transformers.models.fsmt.configuration_fsmt import FSMTConfig
from transformers.optimization import (
    Adafactor,
    AdamW,
    get_constant_schedule,
    get_constant_schedule_with_warmup,
    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.trainer_pt_utils import get_tpu_sampler
from transformers.training_args import ParallelMode
from transformers.utils import is_torch_tpu_available


if is_fairscale_available():
 from fairscale.optim import OSS


_UpperCamelCase 			=		logging.get_logger(__name__)

_UpperCamelCase 			=		{
    "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,
    "constant": get_constant_schedule,
    "constant_w_warmup": get_constant_schedule_with_warmup,
}







class   __lowercase      (_UpperCAmelCase ):
 def __init__(							self      ,							A_=None      ,							A_=None      ,							*A_      ,							**A_					)					->Optional[int]:



       '''simple docstring'''





       super().__init__(*lowercase_      ,							**lowercase_					)

       if config is None:
             assert isinstance(self.model      ,							lowercase_					), (
                 "If no `config` is passed the model to be trained has to be of type `PreTrainedModel`, but is"
                 f""" {self.model.__class__}"""
             )
             __lowerCAmelCase					:						Any					=					self.model.config
       else:
             __lowerCAmelCase					:						Tuple					=					config

       __lowerCAmelCase					:						Tuple					=					data_args
       __lowerCAmelCase					:						Optional[Any]					=					self.config.tgt_vocab_size if isinstance(self.config      ,							lowercase_					) else self.config.vocab_size

       if self.args.label_smoothing != 0 or (self.data_args is not None and self.data_args.ignore_pad_token_for_loss):
             assert self.config.pad_token_id is not None, (
                 "Make sure that `config.pad_token_id` is correcly defined when ignoring `pad_token` for loss"
                 " calculation or doing label smoothing."
             )

       if self.config.pad_token_id is None and self.config.eos_token_id is not None:
             logger.warning(
                 f"""The `config.pad_token_id` is `None`. Using `config.eos_token_id` = {self.config.eos_token_id} for"""
                 ''' padding..'''					)

       if self.args.label_smoothing == 0:
             __lowerCAmelCase					:						List[Any]					=					torch.nn.CrossEntropyLoss(ignore_index=self.config.pad_token_id					)
       else:
             # dynamically import label_smoothed_nll_loss
             from utils import label_smoothed_nll_loss

             __lowerCAmelCase					:						Optional[Any]					=					label_smoothed_nll_loss
 def 	UpperCamelCase__  (							self      ,							A_					)					->Tuple:



       '''simple docstring'''





       if self.optimizer is None:
             __lowerCAmelCase					:						List[str]					=					["""bias""", """LayerNorm.weight"""]
             __lowerCAmelCase					:						List[Any]					=					[
                 {
                     """params""": [p for n, p in self.model.named_parameters() if not any(nd in n for nd in no_decay					)],
                     """weight_decay""": self.args.weight_decay,
                 },
                 {
                     """params""": [p for n, p in self.model.named_parameters() if any(nd in n for nd in no_decay					)],
                     """weight_decay""": 0.0,
                 },
             ]
             __lowerCAmelCase					:						Tuple					=					Adafactor if self.args.adafactor else AdamW
             if self.args.adafactor:
                   __lowerCAmelCase					:						str					=					Adafactor
                   __lowerCAmelCase					:						List[str]					=					{"""scale_parameter""": False, """relative_step""": False}
             else:
                   __lowerCAmelCase					:						List[str]					=					AdamW
                   __lowerCAmelCase					:						Tuple					=					{
                       """betas""": (self.args.adam_betaa, self.args.adam_betaa),
                       """eps""": self.args.adam_epsilon,
                   }
             __lowerCAmelCase					:						Optional[int]					=					self.args.learning_rate
             if self.sharded_ddp:
                   __lowerCAmelCase					:						Dict					=					OSS(
                       params=lowercase_      ,							optim=lowercase_      ,							**lowercase_      ,							)
             else:
                   __lowerCAmelCase					:						Any					=					optimizer_cls(lowercase_      ,							**lowercase_					)

       if self.lr_scheduler is None:
             __lowerCAmelCase					:						List[str]					=					self._get_lr_scheduler(lowercase_					)
       else:  # ignoring --lr_scheduler
             logger.warning('''scheduler is passed to `Seq2SeqTrainer`, `--lr_scheduler` arg is ignored.'''					)
 def 	UpperCamelCase__  (							self      ,							A_					)					->Any:



       '''simple docstring'''





       __lowerCAmelCase					:						Any					=					arg_to_scheduler[self.args.lr_scheduler]
       if self.args.lr_scheduler == "constant":
             __lowerCAmelCase					:						Optional[int]					=					schedule_func(self.optimizer					)
       elif self.args.lr_scheduler == "constant_w_warmup":
             __lowerCAmelCase					:						Tuple					=					schedule_func(self.optimizer      ,							num_warmup_steps=self.args.warmup_steps					)
       else:
             __lowerCAmelCase					:						Optional[int]					=					schedule_func(
                 self.optimizer      ,							num_warmup_steps=self.args.warmup_steps      ,							num_training_steps=lowercase_					)
       return scheduler
 def 	UpperCamelCase__  (							self					)					->Tuple:



       '''simple docstring'''





       if isinstance(self.train_dataset      ,							torch.utils.data.IterableDataset					):
             return None
       elif is_torch_tpu_available():
             return get_tpu_sampler(self.train_dataset					)
       else:
             if self.args.sortish_sampler:
                   self.train_dataset.make_sortish_sampler(
                       self.args.per_device_train_batch_size      ,							distributed=(self.args.parallel_mode == ParallelMode.DISTRIBUTED)      ,							)

             return (
                 RandomSampler(self.train_dataset					)
                 if self.args.local_rank == -1
                 else DistributedSampler(self.train_dataset					)
             )
 def 	UpperCamelCase__  (							self      ,							A_      ,							A_      ,							A_					)					->Optional[int]:



       '''simple docstring'''





       if self.args.label_smoothing == 0:
             if self.data_args is not None and self.data_args.ignore_pad_token_for_loss:
                   # force training to ignore pad token
                   __lowerCAmelCase					:						Optional[Any]					=					model(**lowercase_      ,							use_cache=lowercase_					)[0]
                   __lowerCAmelCase					:						List[Any]					=					self.loss_fn(logits.view(-1      ,							logits.shape[-1]					)      ,							labels.view(-1					)					)
             else:
                   # compute usual loss via models
                   __lowerCAmelCase					:						Dict					=					model(**lowercase_      ,							labels=lowercase_      ,							use_cache=lowercase_					)[:2]
       else:
             # compute label smoothed loss
             __lowerCAmelCase					:						Optional[int]					=					model(**lowercase_      ,							use_cache=lowercase_					)[0]
             __lowerCAmelCase					:						Optional[Any]					=					torch.nn.functional.log_softmax(lowercase_      ,							dim=-1					)
             __lowerCAmelCase					:						Any					=					self.loss_fn(lowercase_      ,							lowercase_      ,							self.args.label_smoothing      ,							ignore_index=self.config.pad_token_id					)
       return loss, logits
 def 	UpperCamelCase__  (							self      ,							A_      ,							A_					)					->Dict:



       '''simple docstring'''





       __lowerCAmelCase					:						List[Any]					=					inputs.pop('''labels'''					)
       __lowerCAmelCase					:						Optional[Any]					=					self._compute_loss(lowercase_      ,							lowercase_      ,							lowercase_					)
       return loss
 def 	UpperCamelCase__  (							self      ,							A_      ,							A_      ,							A_      ,							A_ = None      ,							)					->str:



       '''simple docstring'''





       __lowerCAmelCase					:						Optional[int]					=					self._prepare_inputs(lowercase_					)

       __lowerCAmelCase					:						List[Any]					=					{
           """max_length""": self.data_args.val_max_target_length
           if self.data_args is not None
           else self.config.max_length,
           """num_beams""": self.data_args.eval_beams if self.data_args is not None else self.config.num_beams,
       }

       if self.args.predict_with_generate and not self.args.prediction_loss_only:
             __lowerCAmelCase					:						Union[str, Any]					=					self.model.generate(
                 inputs['''input_ids''']      ,							attention_mask=inputs['''attention_mask''']      ,							**lowercase_      ,							)
             # in case the batch is shorter than max length, the output should be padded
             if generated_tokens.shape[-1] < gen_kwargs["max_length"]:
                   __lowerCAmelCase					:						Dict					=					self._pad_tensors_to_max_len(lowercase_      ,							gen_kwargs['''max_length''']					)

       __lowerCAmelCase					:						str					=					inputs.pop('''labels'''					)
       with torch.no_grad():
             # compute loss on predict data
             __lowerCAmelCase					:						Union[str, Any]					=					self._compute_loss(lowercase_      ,							lowercase_      ,							lowercase_					)

       __lowerCAmelCase					:						List[str]					=					loss.mean().detach()
       if self.args.prediction_loss_only:
             return (loss, None, None)

       __lowerCAmelCase					:						Optional[int]					=					generated_tokens if self.args.predict_with_generate else logits

       if labels.shape[-1] < gen_kwargs["max_length"]:
             __lowerCAmelCase					:						Any					=					self._pad_tensors_to_max_len(lowercase_      ,							gen_kwargs['''max_length''']					)

       return (loss, logits, labels)





 def 	UpperCamelCase__  (							self      ,							A_      ,							A_					)					->str:



       '''simple docstring'''





       __lowerCAmelCase					:						Optional[Any]					=					self.config.pad_token_id if self.config.pad_token_id is not None else self.config.eos_token_id

       if pad_token_id is None:
             raise ValueError(
                 '''Make sure that either `config.pad_token_id` or `config.eos_token_id` is defined if tensor has to be'''
                 f""" padded to `max_length`={max_length}"""					)

       __lowerCAmelCase					:						List[Any]					=					pad_token_id * torch.ones(
           (tensor.shape[0], max_length)      ,							dtype=tensor.dtype      ,							device=tensor.device					)
       __lowerCAmelCase					:						Dict					=					tensor
       return padded_tensor



'''simple docstring'''




import math
import unittest




def 							lowerCamelCase	(      UpperCAmelCase__       :					int )			->      bool:
							assert isinstance(UpperCAmelCase__      ,      UpperCAmelCase__ ) and (
							    number >= 0
							), "'number' must been an int and positive"

							if 1 < number < 4:
														# 2 and 3 are primes
														return True
							elif number < 2 or number % 2 == 0 or number % 3 == 0:
														# Negatives, 0, 1, all even numbers, all multiples of 3 are not primes
														return False

							# All primes number are in format of 6k +/- 1
							for i in range(5      ,      int(math.sqrt(UpperCAmelCase__ ) + 1 )      ,      6 ):
														if number % i == 0 or number % (i + 2) == 0:
																					return False
							return True





class     __magic_name__						(    unittest.TestCase):







			def 	SCREAMING_SNAKE_CASE_     (	self				:  Dict    ):
										self.assertTrue(is_prime(2    )    )
										self.assertTrue(is_prime(3    )    )
										self.assertTrue(is_prime(5    )    )
										self.assertTrue(is_prime(7    )    )
										self.assertTrue(is_prime(11    )    )
										self.assertTrue(is_prime(13    )    )
										self.assertTrue(is_prime(17    )    )
										self.assertTrue(is_prime(19    )    )
										self.assertTrue(is_prime(23    )    )
										self.assertTrue(is_prime(29    )    )







			def 	SCREAMING_SNAKE_CASE_     (	self				:  Optional[Any]    ):
										with self.assertRaises(lowercase_    ):
																	is_prime(-19    )
										self.assertFalse(
										    is_prime(0    )				,			"""Zero doesn't have any positive factors, primes must have exactly two."""				,			)
										self.assertFalse(
										    is_prime(1    )				,			"""One only has 1 positive factor, primes must have exactly two."""				,			)
										self.assertFalse(is_prime(2 * 2    )    )
										self.assertFalse(is_prime(2 * 3    )    )
										self.assertFalse(is_prime(3 * 3    )    )
										self.assertFalse(is_prime(3 * 5    )    )
										self.assertFalse(is_prime(3 * 5 * 7    )    )


if __name__ == "__main__":
						unittest.main()





