






import unittest

from transformers import GPTSwaTokenizer
from transformers.testing_utils import get_tests_dir, require_sentencepiece, require_tokenizers, slow

from ...test_tokenization_common import TokenizerTesterMixin


A__	:			Union[str, Any]			   = get_tests_dir('fixtures/test_sentencepiece_with_bytefallback.model')
@require_sentencepiece
@require_tokenizers
class       _UpperCAmelCase					(       A__		,unittest.TestCase	):

				"""simple docstring"""




				lowercase__											=       GPTSwaTokenizer
				lowercase__											=       False
				lowercase__											=       True
				lowercase__											=       False




				def 				lowercase__   (     self						:    str  ):

								'''simple docstring'''
								super().setUp()

								# We have a SentencePiece fixture for testing
								lowercase__    =			GPTSwaTokenizer(lowerCamelCase,  eos_token='''<unk>''',  bos_token='''<unk>''',  pad_token='''<unk>'''  )

								tokenizer.save_pretrained(self.tmpdirname  )




				def 				lowercase__   (     self						:    Union[str, Any],  lowerCamelCase						:    Optional[Any]  ):

								'''simple docstring'''
								lowercase__    =			'''This is a test'''
								lowercase__    =			'''This is a test'''
								return input_text, output_text




				def 				lowercase__   (     self						:    Optional[int]  ):

								'''simple docstring'''
								lowercase__    =			'''<s>'''
								lowercase__    =			1

								self.assertEqual(self.get_tokenizer()._convert_token_to_id(lowerCamelCase  ),  lowerCamelCase  )
								self.assertEqual(self.get_tokenizer()._convert_id_to_token(lowerCamelCase  ),  lowerCamelCase  )




				def 				lowercase__   (     self						:    Optional[int]  ):

								'''simple docstring'''
								lowercase__    =			list(self.get_tokenizer().get_vocab().keys()  )

								self.assertEqual(vocab_keys[0],  '''<unk>'''  )
								self.assertEqual(vocab_keys[1],  '''<s>'''  )
								self.assertEqual(vocab_keys[-1],  '''j'''  )
								self.assertEqual(len(lowerCamelCase  ),  2_000  )




				def 				lowercase__   (     self						:    Optional[int]  ):

								'''simple docstring'''
								self.assertEqual(self.get_tokenizer().vocab_size,  2_000  )




				def 				lowercase__   (     self						:    Tuple  ):

								'''simple docstring'''
								lowercase__    =			GPTSwaTokenizer(lowerCamelCase  )

								lowercase__    =			tokenizer.tokenize('''This is a test'''  )
								self.assertListEqual(lowerCamelCase,  ['''▁This''', '''▁is''', '''▁a''', '''▁t''', '''est''']  )

								self.assertListEqual(tokenizer.convert_tokens_to_ids(lowerCamelCase  ),  [465, 287, 265, 631, 842]  )

								lowercase__    =			tokenizer.tokenize('''I was born in 92000, and this is falsé.'''  )
								# fmt: off
								self.assertListEqual(
								    lowerCamelCase,  ['''▁I''', '''▁was''', '''▁bor''', '''n''', '''▁in''', '''▁''', '''<0x39>''', '''2''', '''0''', '''0''', '''0''', ''',''', '''▁and''', '''▁this''', '''▁is''', '''▁f''', '''al''', '''s''', '''<0xC3>''', '''<0xA9>''', '''.'''],  )
								# fmt: on

								lowercase__    =			tokenizer.convert_tokens_to_ids(lowerCamelCase  )
								self.assertListEqual(
								    lowerCamelCase,  [262, 272, 1_525, 286, 271, 268, 60, 916, 633, 633, 633, 259, 266, 301, 287, 384, 367, 263, 198, 172, 260],  )

								lowercase__    =			tokenizer.convert_ids_to_tokens(lowerCamelCase  )
								# fmt: off
								self.assertListEqual(
								    lowerCamelCase,  ['''▁I''', '''▁was''', '''▁bor''', '''n''', '''▁in''', '''▁''', '''<0x39>''', '''2''', '''0''', '''0''', '''0''', ''',''', '''▁and''', '''▁this''', '''▁is''', '''▁f''', '''al''', '''s''', '''<0xC3>''', '''<0xA9>''', '''.''']  )
								# fmt: on




				def 				lowercase__   (     self						:    Tuple  ):

								'''simple docstring'''
								lowercase__    =			GPTSwaTokenizer(lowerCamelCase  )
								lowercase__    =			['''This is a test''', '''I was born in 92000, and this is falsé.''']
								lowercase__    =			[
								    [465, 287, 265, 631, 842],
								    [262, 272, 1_525, 286, 271, 268, 60, 916, 633, 633, 633, 259, 266, 301, 287, 384, 367, 263, 198, 172, 260],
								]

								# Test that encode_fast returns the same as tokenize + convert_tokens_to_ids
								for text, expected_ids in zip(lowerCamelCase,  lowerCamelCase  ):
												self.assertListEqual(tokenizer.encode_fast(lowerCamelCase  ),  lowerCamelCase  )

								# Test that decode_fast returns the input text
								for text, token_ids in zip(lowerCamelCase,  lowerCamelCase  ):
												self.assertEqual(tokenizer.decode_fast(lowerCamelCase  ),  lowerCamelCase  )




				@slow
				def 				lowercase__   (     self						:    List[str]  ):

								'''simple docstring'''
								lowercase__    =			[
								    '''<|python|>def fibonacci(n)\n    if n < 0:\n        print(\'Incorrect input\')''',
								    '''Hey there, how are you doing this fine day?''',
								    '''This is a text with a trailing spaces followed by a dot     .''',
								    '''Häj sväjs lillebrör! =)''',
								    '''Det är inget fel på Mr. Cool''',
								]

								# fmt: off
								lowercase__    =			{'''input_ids''': [[63_423, 5, 6_811, 14_954, 282, 816, 3_821, 63_466, 63_425, 63_462, 18, 63_978, 678, 301, 1_320, 63_423, 63_455, 63_458, 18, 63_982, 4_246, 3_940, 1_901, 47_789, 5_547, 18_994], [19_630, 1_100, 63_446, 1_342, 633, 544, 4_488, 593, 5_102, 2_416, 63_495, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [1_652, 428, 268, 1_936, 515, 268, 58_593, 22_413, 9_106, 546, 268, 33_213, 63_979, 698, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [55_130, 63_450, 924, 63_449, 2_249, 4_062, 1_558, 318, 63_504, 21_498, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [509, 377, 2_827, 2_559, 332, 6_575, 63_443, 26_801, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]], '''token_type_ids''': [[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]], '''attention_mask''': [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]]}
								# fmt: on
								self.tokenizer_integration_test_util(
								    expected_encoding=lowerCamelCase,  model_name='''AI-Sweden/gpt-sw3-126m''',  sequences=lowerCamelCase,  )












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


A__	:			Tuple			   = logging.get_logger(__name__)

A__	:			int			   = {
    '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       _UpperCAmelCase					(       A__	):

				"""simple docstring"""




				def __init__(     self						:    Optional[int],  lowerCamelCase						:    int=None,  lowerCamelCase						:    int=None,  *lowerCamelCase						:    List[Any],  **lowerCamelCase						:    Any  ):

								'''simple docstring'''
								super().__init__(*lowerCamelCase,  **lowerCamelCase  )

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

								lowercase__    =			data_args
								lowercase__    =			self.config.tgt_vocab_size if isinstance(self.config,  lowerCamelCase  ) 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:
												lowercase__    =			torch.nn.CrossEntropyLoss(ignore_index=self.config.pad_token_id  )
								else:
												# dynamically import label_smoothed_nll_loss
												from utils import label_smoothed_nll_loss

												lowercase__    =			label_smoothed_nll_loss




				def 				lowercase__   (     self						:    List[Any],  lowerCamelCase						:    int  ):

								'''simple docstring'''
								if self.optimizer is None:
												lowercase__    =			['''bias''', '''LayerNorm.weight''']
												lowercase__    =			[
												    {
												        '''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,
												    },
												]
												lowercase__    =			Adafactor if self.args.adafactor else AdamW
												if self.args.adafactor:
																lowercase__    =			Adafactor
																lowercase__    =			{'''scale_parameter''': False, '''relative_step''': False}
												else:
																lowercase__    =			AdamW
																lowercase__    =			{
																    '''betas''': (self.args.adam_betaa, self.args.adam_betaa),
																    '''eps''': self.args.adam_epsilon,
																}
												lowercase__    =			self.args.learning_rate
												if self.sharded_ddp:
																lowercase__    =			OSS(
																    params=lowerCamelCase,  optim=lowerCamelCase,  **lowerCamelCase,  )
												else:
																lowercase__    =			optimizer_cls(lowerCamelCase,  **lowerCamelCase  )

								if self.lr_scheduler is None:
												lowercase__    =			self._get_lr_scheduler(lowerCamelCase  )
								else:  # ignoring --lr_scheduler
												logger.warning('''scheduler is passed to `Seq2SeqTrainer`, `--lr_scheduler` arg is ignored.'''  )




				def 				lowercase__   (     self						:    List[str],  lowerCamelCase						:    Optional[int]  ):

								'''simple docstring'''
								lowercase__    =			arg_to_scheduler[self.args.lr_scheduler]
								if self.args.lr_scheduler == "constant":
												lowercase__    =			schedule_func(self.optimizer  )
								elif self.args.lr_scheduler == "constant_w_warmup":
												lowercase__    =			schedule_func(self.optimizer,  num_warmup_steps=self.args.warmup_steps  )
								else:
												lowercase__    =			schedule_func(
												    self.optimizer,  num_warmup_steps=self.args.warmup_steps,  num_training_steps=lowerCamelCase  )
								return scheduler




				def 				lowercase__   (     self						:    List[Any]  ):

								'''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 				lowercase__   (     self						:    Any,  lowerCamelCase						:    List[Any],  lowerCamelCase						:    str,  lowerCamelCase						:    Union[str, Any]  ):

								'''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
																lowercase__    =			model(**lowerCamelCase,  use_cache=lowerCamelCase  )[0]
																lowercase__    =			self.loss_fn(logits.view(-1,  logits.shape[-1]  ),  labels.view(-1  )  )
												else:
																# compute usual loss via models
																lowercase__ ,						lowercase__    =			model(**lowerCamelCase,  labels=lowerCamelCase,  use_cache=lowerCamelCase  )[:2]
								else:
												# compute label smoothed loss
												lowercase__    =			model(**lowerCamelCase,  use_cache=lowerCamelCase  )[0]
												lowercase__    =			torch.nn.functional.log_softmax(lowerCamelCase,  dim=-1  )
												lowercase__ ,						lowercase__    =			self.loss_fn(lowerCamelCase,  lowerCamelCase,  self.args.label_smoothing,  ignore_index=self.config.pad_token_id  )
								return loss, logits




				def 				lowercase__   (     self						:    List[str],  lowerCamelCase						:    Optional[Any],  lowerCamelCase						:    Tuple  ):

								'''simple docstring'''
								lowercase__    =			inputs.pop('''labels'''  )
								lowercase__ ,						lowercase__    =			self._compute_loss(lowerCamelCase,  lowerCamelCase,  lowerCamelCase  )
								return loss




				def 				lowercase__   (     self						:    str,  lowerCamelCase						:    nn.Module,  lowerCamelCase						:    Dict[str, Union[torch.Tensor, Any]],  lowerCamelCase						:    bool,  lowerCamelCase						:    Optional[List[str]] = None,  ):

								'''simple docstring'''
								lowercase__    =			self._prepare_inputs(lowerCamelCase  )

								lowercase__    =			{
								    '''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:
												lowercase__    =			self.model.generate(
												    inputs['''input_ids'''],  attention_mask=inputs['''attention_mask'''],  **lowerCamelCase,  )
												# in case the batch is shorter than max length, the output should be padded
												if generated_tokens.shape[-1] < gen_kwargs["max_length"]:
																lowercase__    =			self._pad_tensors_to_max_len(lowerCamelCase,  gen_kwargs['''max_length''']  )

								lowercase__    =			inputs.pop('''labels'''  )
								with torch.no_grad():
												# compute loss on predict data
												lowercase__ ,						lowercase__    =			self._compute_loss(lowerCamelCase,  lowerCamelCase,  lowerCamelCase  )

								lowercase__    =			loss.mean().detach()
								if self.args.prediction_loss_only:
												return (loss, None, None)

								lowercase__    =			generated_tokens if self.args.predict_with_generate else logits

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

								return (loss, logits, labels)




				def 				lowercase__   (     self						:    List[Any],  lowerCamelCase						:    str,  lowerCamelCase						:    Any  ):

								'''simple docstring'''
								# If PAD token is not defined at least EOS token has to be defined
								lowercase__    =			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}"""  )

								lowercase__    =			pad_token_id * torch.ones(
								    (tensor.shape[0], max_length),  dtype=tensor.dtype,  device=tensor.device  )
								lowercase__    =			tensor
								return padded_tensor




