
import shutil
import tempfile
import unittest
from unittest.mock import patch

from transformers import (
    DefaultFlowCallback,
    IntervalStrategy,
    PrinterCallback,
    ProgressCallback,
    Trainer,
    TrainerCallback,
    TrainingArguments,
    is_torch_available,
)
from transformers.testing_utils import require_torch


if is_torch_available():
   from transformers.trainer import DEFAULT_CALLBACKS

   from .test_trainer import RegressionDataset, RegressionModelConfig, RegressionPreTrainedModel







class   lowerCamelCase	(						A_						):





     def __init__(self   :					int     )		->		str:
         snake_case				  =      []





     def 	UpperCAmelCase(self   :					Any       ,					_A   :					int       ,					_A   :					Optional[int]       ,					_A   :					Dict       ,					**_A   :					List[str]     )		->		int:
         self.events.append("on_init_end"     )





     def 	UpperCAmelCase(self   :					List[str]       ,					_A   :					List[Any]       ,					_A   :					List[Any]       ,					_A   :					List[Any]       ,					**_A   :					Optional[int]     )		->		Tuple:
         self.events.append("on_train_begin"     )





     def 	UpperCAmelCase(self   :					Union[str, Any]       ,					_A   :					Any       ,					_A   :					Tuple       ,					_A   :					Dict       ,					**_A   :					str     )		->		Any:
         self.events.append("on_train_end"     )





     def 	UpperCAmelCase(self   :					Dict       ,					_A   :					Tuple       ,					_A   :					Dict       ,					_A   :					List[Any]       ,					**_A   :					Tuple     )		->		List[Any]:
         self.events.append("on_epoch_begin"     )





     def 	UpperCAmelCase(self   :					Dict       ,					_A   :					Dict       ,					_A   :					Dict       ,					_A   :					Optional[int]       ,					**_A   :					Optional[int]     )		->		Optional[int]:
         self.events.append("on_epoch_end"     )





     def 	UpperCAmelCase(self   :					str       ,					_A   :					Any       ,					_A   :					Optional[int]       ,					_A   :					List[Any]       ,					**_A   :					List[Any]     )		->		Optional[Any]:
         self.events.append("on_step_begin"     )





     def 	UpperCAmelCase(self   :					int       ,					_A   :					List[str]       ,					_A   :					Union[str, Any]       ,					_A   :					Optional[Any]       ,					**_A   :					List[str]     )		->		str:
         self.events.append("on_step_end"     )





     def 	UpperCAmelCase(self   :					Optional[Any]       ,					_A   :					Tuple       ,					_A   :					Dict       ,					_A   :					List[str]       ,					**_A   :					List[Any]     )		->		Dict:
         self.events.append("on_evaluate"     )





     def 	UpperCAmelCase(self   :					Optional[int]       ,					_A   :					Union[str, Any]       ,					_A   :					Any       ,					_A   :					str       ,					**_A   :					str     )		->		Dict:
         self.events.append("on_predict"     )





     def 	UpperCAmelCase(self   :					int       ,					_A   :					List[Any]       ,					_A   :					Dict       ,					_A   :					Optional[int]       ,					**_A   :					Dict     )		->		int:
         self.events.append("on_save"     )





     def 	UpperCAmelCase(self   :					int       ,					_A   :					str       ,					_A   :					Union[str, Any]       ,					_A   :					Tuple       ,					**_A   :					Tuple     )		->		int:
         self.events.append("on_log"     )





     def 	UpperCAmelCase(self   :					List[Any]       ,					_A   :					Tuple       ,					_A   :					List[Any]       ,					_A   :					int       ,					**_A   :					Dict     )		->		Tuple:
         self.events.append("on_prediction_step"     )







@require_torch
class   lowerCamelCase	(						unittest.TestCase						):





     def 	UpperCAmelCase(self   :					List[str]     )		->		Optional[Any]:
         snake_case				  =      tempfile.mkdtemp()





     def 	UpperCAmelCase(self   :					int     )		->		List[Any]:
         shutil.rmtree(self.output_dir     )





     def 	UpperCAmelCase(self   :					Union[str, Any]       ,					_A   :					List[str]=0       ,					_A   :					Any=0       ,					_A   :					Optional[int]=6_4       ,					_A   :					List[str]=6_4       ,					_A   :					int=None       ,					_A   :					Union[str, Any]=False       ,					**_A   :					Tuple     )		->		List[Any]:
         # disable_tqdm in TrainingArguments has a flaky default since it depends on the level of logging. We make sure
         # its set to False since the tests later on depend on its value.
         snake_case				  =      RegressionDataset(length=_A     )
         snake_case				  =      RegressionDataset(length=_A     )
         snake_case				  =      RegressionModelConfig(a=_A       ,					b=_A     )
         snake_case				  =      RegressionPreTrainedModel(_A     )

         snake_case				  =      TrainingArguments(self.output_dir       ,					disable_tqdm=_A       ,					report_to=[]       ,					**_A     )
         return Trainer(
             _A       ,					_A       ,					train_dataset=_A       ,					eval_dataset=_A       ,					callbacks=_A       ,					)





     def 	UpperCAmelCase(self   :					Tuple       ,					_A   :					Optional[Any]       ,					_A   :					List[str]     )		->		str:
         self.assertEqual(len(_A     )       ,					len(_A     )     )

         # Order doesn't matter
         snake_case				  =      sorted(_A       ,					key=lambda _A     : cb.__name__ if isinstance(_A       ,					_A     ) else cb.__class__.__name__     )
         snake_case				  =      sorted(_A       ,					key=lambda _A     : cb.__name__ if isinstance(_A       ,					_A     ) else cb.__class__.__name__     )

         for cba, cba in zip(_A       ,					_A     ):
             if isinstance(_A       ,					_A     ) and isinstance(_A       ,					_A     ):
                 self.assertEqual(_A       ,					_A     )
             elif isinstance(_A       ,					_A     ) and not isinstance(_A       ,					_A     ):
                 self.assertEqual(_A       ,					cba.__class__     )
             elif not isinstance(_A       ,					_A     ) and isinstance(_A       ,					_A     ):
                 self.assertEqual(cba.__class__       ,					_A     )
             else:
                 self.assertEqual(_A       ,					_A     )





     def 	UpperCAmelCase(self   :					Tuple       ,					_A   :					str     )		->		List[Any]:
         snake_case				  =      ["on_init_end", "on_train_begin"]
         snake_case				  =      0
         snake_case				  =      len(trainer.get_eval_dataloader()     )
         snake_case				  =      ["on_prediction_step"] * len(trainer.get_eval_dataloader()     ) + ["on_log", "on_evaluate"]
         for _ in range(trainer.state.num_train_epochs     ):
             expected_events.append("on_epoch_begin"     )
             for _ in range(_A     ):
                 step += 1
                 expected_events += ["on_step_begin", "on_step_end"]
                 if step % trainer.args.logging_steps == 0:
                     expected_events.append("on_log"     )
                 if trainer.args.evaluation_strategy == IntervalStrategy.STEPS and step % trainer.args.eval_steps == 0:
                     expected_events += evaluation_events.copy()
                 if step % trainer.args.save_steps == 0:
                     expected_events.append("on_save"     )
             expected_events.append("on_epoch_end"     )
             if trainer.args.evaluation_strategy == IntervalStrategy.EPOCH:
                 expected_events += evaluation_events.copy()
         expected_events += ["on_log", "on_train_end"]
         return expected_events





     def 	UpperCAmelCase(self   :					Optional[Any]     )		->		Optional[Any]:
         snake_case				  =      self.get_trainer()
         snake_case				  =      DEFAULT_CALLBACKS.copy() + [ProgressCallback]
         self.check_callbacks_equality(trainer.callback_handler.callbacks       ,					_A     )

         # Callbacks passed at init are added to the default callbacks
         snake_case				  =      self.get_trainer(callbacks=[MyTestTrainerCallback]     )
         expected_callbacks.append(_A     )
         self.check_callbacks_equality(trainer.callback_handler.callbacks       ,					_A     )

         # TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
         snake_case				  =      self.get_trainer(disable_tqdm=_A     )
         snake_case				  =      DEFAULT_CALLBACKS.copy() + [PrinterCallback]
         self.check_callbacks_equality(trainer.callback_handler.callbacks       ,					_A     )





     def 	UpperCAmelCase(self   :					Optional[int]     )		->		Union[str, Any]:
         snake_case				  =      DEFAULT_CALLBACKS.copy() + [ProgressCallback]
         snake_case				  =      self.get_trainer()

         # We can add, pop, or remove by class name
         trainer.remove_callback(_A     )
         expected_callbacks.remove(_A     )
         self.check_callbacks_equality(trainer.callback_handler.callbacks       ,					_A     )

         snake_case				  =      self.get_trainer()
         snake_case				  =      trainer.pop_callback(_A     )
         self.assertEqual(cb.__class__       ,					_A     )
         self.check_callbacks_equality(trainer.callback_handler.callbacks       ,					_A     )

         trainer.add_callback(_A     )
         expected_callbacks.insert(0       ,					_A     )
         self.check_callbacks_equality(trainer.callback_handler.callbacks       ,					_A     )

         # We can also add, pop, or remove by instance
         snake_case				  =      self.get_trainer()
         snake_case				  =      trainer.callback_handler.callbacks[0]
         trainer.remove_callback(_A     )
         expected_callbacks.remove(_A     )
         self.check_callbacks_equality(trainer.callback_handler.callbacks       ,					_A     )

         snake_case				  =      self.get_trainer()
         snake_case				  =      trainer.callback_handler.callbacks[0]
         snake_case				  =      trainer.pop_callback(_A     )
         self.assertEqual(_A       ,					_A     )
         self.check_callbacks_equality(trainer.callback_handler.callbacks       ,					_A     )

         trainer.add_callback(_A     )
         expected_callbacks.insert(0       ,					_A     )
         self.check_callbacks_equality(trainer.callback_handler.callbacks       ,					_A     )





     def 	UpperCAmelCase(self   :					Tuple     )		->		List[Any]:
         import warnings

         # XXX: for now ignore scatter_gather warnings in this test since it's not relevant to what's being tested
         warnings.simplefilter(action="ignore"       ,					category=_A     )

         snake_case				  =      self.get_trainer(callbacks=[MyTestTrainerCallback]     )
         trainer.train()
         snake_case				  =      trainer.callback_handler.callbacks[-2].events
         self.assertEqual(_A       ,					self.get_expected_events(_A     )     )

         # Independent log/save/eval
         snake_case				  =      self.get_trainer(callbacks=[MyTestTrainerCallback]       ,					logging_steps=5     )
         trainer.train()
         snake_case				  =      trainer.callback_handler.callbacks[-2].events
         self.assertEqual(_A       ,					self.get_expected_events(_A     )     )

         snake_case				  =      self.get_trainer(callbacks=[MyTestTrainerCallback]       ,					save_steps=5     )
         trainer.train()
         snake_case				  =      trainer.callback_handler.callbacks[-2].events
         self.assertEqual(_A       ,					self.get_expected_events(_A     )     )

         snake_case				  =      self.get_trainer(callbacks=[MyTestTrainerCallback]       ,					eval_steps=5       ,					evaluation_strategy="steps"     )
         trainer.train()
         snake_case				  =      trainer.callback_handler.callbacks[-2].events
         self.assertEqual(_A       ,					self.get_expected_events(_A     )     )

         snake_case				  =      self.get_trainer(callbacks=[MyTestTrainerCallback]       ,					evaluation_strategy="epoch"     )
         trainer.train()
         snake_case				  =      trainer.callback_handler.callbacks[-2].events
         self.assertEqual(_A       ,					self.get_expected_events(_A     )     )

         # A bit of everything
         snake_case				  =      self.get_trainer(
             callbacks=[MyTestTrainerCallback]       ,					logging_steps=3       ,					save_steps=1_0       ,					eval_steps=5       ,					evaluation_strategy="steps"       ,					)
         trainer.train()
         snake_case				  =      trainer.callback_handler.callbacks[-2].events
         self.assertEqual(_A       ,					self.get_expected_events(_A     )     )

         # warning should be emitted for duplicated callbacks
         with patch("transformers.trainer_callback.logger.warning"     ) as warn_mock:
             snake_case				  =      self.get_trainer(
                 callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]       ,					)
             assert str(_A     ) in warn_mock.call_args[0][0]



import os
import unittest

from transformers.models.phobert.tokenization_phobert import VOCAB_FILES_NAMES, PhobertTokenizer

from ...test_tokenization_common import TokenizerTesterMixin







class   lowerCamelCase	(						A_							,				unittest.TestCase						):
     UpperCAmelCase__       :							str     =	PhobertTokenizer
     UpperCAmelCase__       :							List[Any]     =	False





     def 	UpperCAmelCase(self   :					Union[str, Any]     )		->		List[str]:
         super().setUp()

         # Adapted from Sennrich et al. 2015 and https://github.com/rsennrich/subword-nmt
         snake_case				  =      ["T@@", "i", "I", "R@@", "r", "e@@"]
         snake_case				  =      dict(zip(_A       ,					range(len(_A     )     )     )     )
         snake_case				  =      ["#version: 0.2", "l à</w>"]
         snake_case				  =      {"unk_token": "<unk>"}

         snake_case				  =      os.path.join(self.tmpdirname       ,					VOCAB_FILES_NAMES["vocab_file"]     )
         snake_case				  =      os.path.join(self.tmpdirname       ,					VOCAB_FILES_NAMES["merges_file"]     )

         with open(self.vocab_file       ,					"w"       ,					encoding="utf-8"     ) as fp:
             for token in vocab_tokens:
                 fp.write(f'{token} {vocab_tokens[token]}\n'     )
         with open(self.merges_file       ,					"w"       ,					encoding="utf-8"     ) as fp:
             fp.write("\n".join(_A     )     )





     def 	UpperCAmelCase(self   :					Optional[int]       ,					**_A   :					Optional[int]     )		->		Union[str, Any]:
         kwargs.update(self.special_tokens_map     )
         return PhobertTokenizer.from_pretrained(self.tmpdirname       ,					**_A     )





     def 	UpperCAmelCase(self   :					Tuple       ,					_A   :					Optional[Any]     )		->		str:
         snake_case				  =      "Tôi là VinAI Research"
         snake_case				  =      "T<unk> i <unk> <unk> <unk> <unk> <unk> <unk> I Re<unk> e<unk> <unk> <unk> <unk>"
         return input_text, output_text





     def 	UpperCAmelCase(self   :					Any     )		->		Union[str, Any]:
         snake_case				  =      PhobertTokenizer(self.vocab_file       ,					self.merges_file       ,					**self.special_tokens_map     )
         snake_case				  =      "Tôi là VinAI Research"
         snake_case				  =      "T@@ ô@@ i l@@ à V@@ i@@ n@@ A@@ I R@@ e@@ s@@ e@@ a@@ r@@ c@@ h".split()
         snake_case				  =      tokenizer.tokenize(_A     )
         print(_A     )
         self.assertListEqual(_A       ,					_A     )

         snake_case				  =      tokens + [tokenizer.unk_token]

         snake_case				  =      [4, 3, 5, 3, 3, 3, 3, 3, 3, 6, 7, 9, 3, 9, 3, 3, 3, 3, 3]
         self.assertListEqual(tokenizer.convert_tokens_to_ids(_A     )       ,					_A     )

