




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 						_A (  _lowerCamelCase				):






  def __init__(  self		:     Union[str, Any]				)		->   Optional[int]:






       """simple docstring"""







       lowercase :       Optional[int]       =			[]






  def  __a      (  self		:     str				,						_A		:     Tuple				,						_A		:     List[str]				,						_A		:     str				,						**_A		:     List[str]				)		->   int:






       """simple docstring"""







       self.events.append('''on_init_end'''				)






  def  __a      (  self		:     Union[str, Any]				,						_A		:     Optional[Any]				,						_A		:     Dict				,						_A		:     Any				,						**_A		:     Optional[Any]				)		->   Any:






       """simple docstring"""







       self.events.append('''on_train_begin'''				)






  def  __a      (  self		:     Optional[int]				,						_A		:     str				,						_A		:     Dict				,						_A		:     str				,						**_A		:     Tuple				)		->   List[Any]:






       """simple docstring"""







       self.events.append('''on_train_end'''				)






  def  __a      (  self		:     Any				,						_A		:     Union[str, Any]				,						_A		:     Tuple				,						_A		:     int				,						**_A		:     Optional[Any]				)		->   str:






       """simple docstring"""







       self.events.append('''on_epoch_begin'''				)






  def  __a      (  self		:     Optional[Any]				,						_A		:     Dict				,						_A		:     Any				,						_A		:     int				,						**_A		:     Optional[Any]				)		->   str:






       """simple docstring"""







       self.events.append('''on_epoch_end'''				)






  def  __a      (  self		:     List[Any]				,						_A		:     str				,						_A		:     Any				,						_A		:     str				,						**_A		:     List[Any]				)		->   List[str]:






       """simple docstring"""







       self.events.append('''on_step_begin'''				)






  def  __a      (  self		:     List[str]				,						_A		:     Tuple				,						_A		:     List[Any]				,						_A		:     Union[str, Any]				,						**_A		:     List[str]				)		->   Tuple:






       """simple docstring"""







       self.events.append('''on_step_end'''				)






  def  __a      (  self		:     Optional[int]				,						_A		:     Optional[int]				,						_A		:     Dict				,						_A		:     Tuple				,						**_A		:     int				)		->   str:






       """simple docstring"""







       self.events.append('''on_evaluate'''				)






  def  __a      (  self		:     Union[str, Any]				,						_A		:     Any				,						_A		:     Dict				,						_A		:     str				,						**_A		:     int				)		->   Union[str, Any]:






       """simple docstring"""







       self.events.append('''on_predict'''				)






  def  __a      (  self		:     Optional[Any]				,						_A		:     Any				,						_A		:     List[str]				,						_A		:     Optional[int]				,						**_A		:     Tuple				)		->   Tuple:






       """simple docstring"""







       self.events.append('''on_save'''				)






  def  __a      (  self		:     List[Any]				,						_A		:     List[str]				,						_A		:     str				,						_A		:     Union[str, Any]				,						**_A		:     List[Any]				)		->   Any:






       """simple docstring"""







       self.events.append('''on_log'''				)






  def  __a      (  self		:     Optional[int]				,						_A		:     Any				,						_A		:     int				,						_A		:     int				,						**_A		:     str				)		->   int:






       """simple docstring"""







       self.events.append('''on_prediction_step'''				)


@require_torch
class 						_A (  unittest.TestCase				):






  def  __a      (  self		:     Union[str, Any]				)		->   Optional[int]:






       """simple docstring"""







       lowercase :       List[Any]       =			tempfile.mkdtemp()






  def  __a      (  self		:     int				)		->   str:






       """simple docstring"""







       shutil.rmtree(self.output_dir				)






  def  __a      (  self		:     int				,						_A		:     Tuple=0				,						_A		:     Optional[int]=0				,						_A		:     Dict=64				,						_A		:     Union[str, Any]=64				,						_A		:     List[Any]=None				,						_A		:     Any=False				,						**_A		:     List[Any]				)		->   Optional[int]:






       """simple docstring"""







       lowercase :       Union[str, Any]       =			RegressionDataset(length=_A				)
       lowercase :       Optional[int]       =			RegressionDataset(length=_A				)
       lowercase :       Union[str, Any]       =			RegressionModelConfig(a=_A				,						b=_A				)
       lowercase :       List[Any]       =			RegressionPreTrainedModel(_A				)

       lowercase :       Optional[Any]       =			TrainingArguments(self.output_dir				,						disable_tqdm=_A				,						report_to=[]				,						**_A				)
       return Trainer(
           _A				,						_A				,						train_dataset=_A				,						eval_dataset=_A				,						callbacks=_A				,						)






  def  __a      (  self		:     Tuple				,						_A		:     Optional[int]				,						_A		:     int				)		->   Tuple:






       """simple docstring"""







       self.assertEqual(len(_A				)				,						len(_A				)				)

       # Order doesn't matter
       lowercase :       Dict       =			sorted(_A				,						key=lambda _A				: cb.__name__ if isinstance(_A				,						_A				) else cb.__class__.__name__				)
       lowercase :       Any       =			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  __a      (  self		:     List[Any]				,						_A		:     Optional[int]				)		->   int:






       """simple docstring"""







       lowercase :       List[Any]       =			['''on_init_end''', '''on_train_begin''']
       lowercase :       Dict       =			0
       lowercase :       Optional[int]       =			len(trainer.get_eval_dataloader()				)
       lowercase :       Tuple       =			['''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  __a      (  self		:     List[str]				)		->   Tuple:






       """simple docstring"""







       lowercase :       Optional[int]       =			self.get_trainer()
       lowercase :       Union[str, Any]       =			DEFAULT_CALLBACKS.copy() + [ProgressCallback]
       self.check_callbacks_equality(trainer.callback_handler.callbacks				,						_A				)

       # Callbacks passed at init are added to the default callbacks
       lowercase :       List[Any]       =			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
       lowercase :       Optional[Any]       =			self.get_trainer(disable_tqdm=_A				)
       lowercase :       Optional[Any]       =			DEFAULT_CALLBACKS.copy() + [PrinterCallback]
       self.check_callbacks_equality(trainer.callback_handler.callbacks				,						_A				)






  def  __a      (  self		:     Optional[Any]				)		->   Any:






       """simple docstring"""







       lowercase :       List[str]       =			DEFAULT_CALLBACKS.copy() + [ProgressCallback]
       lowercase :       Optional[int]       =			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				)

       lowercase :       List[Any]       =			self.get_trainer()
       lowercase :       str       =			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
       lowercase :       Optional[int]       =			self.get_trainer()
       lowercase :       Optional[int]       =			trainer.callback_handler.callbacks[0]
       trainer.remove_callback(_A				)
       expected_callbacks.remove(_A				)
       self.check_callbacks_equality(trainer.callback_handler.callbacks				,						_A				)

       lowercase :       Optional[int]       =			self.get_trainer()
       lowercase :       List[str]       =			trainer.callback_handler.callbacks[0]
       lowercase :       int       =			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  __a      (  self		:     Any				)		->   Union[str, Any]:






       """simple docstring"""







       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				)

       lowercase :       Dict       =			self.get_trainer(callbacks=[MyTestTrainerCallback]				)
       trainer.train()
       lowercase :       List[str]       =			trainer.callback_handler.callbacks[-2].events
       self.assertEqual(_A				,						self.get_expected_events(_A				)				)

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

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

       lowercase :       List[str]       =			self.get_trainer(callbacks=[MyTestTrainerCallback]				,						eval_steps=5				,						evaluation_strategy='''steps'''				)
       trainer.train()
       lowercase :       Optional[Any]       =			trainer.callback_handler.callbacks[-2].events
       self.assertEqual(_A				,						self.get_expected_events(_A				)				)

       lowercase :       List[Any]       =			self.get_trainer(callbacks=[MyTestTrainerCallback]				,						evaluation_strategy='''epoch'''				)
       trainer.train()
       lowercase :       Dict       =			trainer.callback_handler.callbacks[-2].events
       self.assertEqual(_A				,						self.get_expected_events(_A				)				)

       # A bit of everything
       lowercase :       Dict       =			self.get_trainer(
           callbacks=[MyTestTrainerCallback]				,						logging_steps=3				,						save_steps=10				,						eval_steps=5				,						evaluation_strategy='''steps'''				,						)
       trainer.train()
       lowercase :       Union[str, Any]       =			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:
            lowercase :       List[Any]       =			self.get_trainer(
                callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]				,						)
            assert str(_A				) in warn_mock.call_args[0][0]





class 						_A :  # Public class to implement a graph






  def __init__(  self		:     List[Any]				,						_A		:     int				,						_A		:     int				,						_A		:     list[list[bool]]				)		->   None:






       """simple docstring"""







       lowercase :       Tuple       =			row
       lowercase :       Union[str, Any]       =			col
       lowercase :       int       =			graph






  def  __a      (  self		:     List[Any]				,						_A		:     int				,						_A		:     int				,						_A		:     list[list[bool]]				)		->   bool:






       """simple docstring"""







       return (
           0 <= i < self.ROW
           and 0 <= j < self.COL
           and not visited[i][j]
           and self.graph[i][j]
       )






  def  __a      (  self		:     int				,						_A		:     int				,						_A		:     int				,						_A		:     list[list[bool]]				)		->   None:






       """simple docstring"""







       lowercase :       List[str]       =			[-1, -1, -1, 0, 0, 1, 1, 1]  # Coordinate order
       lowercase :       Dict       =			[-1, 0, 1, -1, 1, -1, 0, 1]
       lowercase :       Dict       =			True  # Make those cells visited
       for k in range(8				):
            if self.is_safe(i + row_nbr[k]				,						j + col_nbr[k]				,						_A				):
                 self.diffs(i + row_nbr[k]				,						j + col_nbr[k]				,						_A				)






  def  __a      (  self		:     List[str]				)		->   int:  # And finally, count all islands.






       """simple docstring"""







       lowercase :       List[str]       =			[[False for j in range(self.COL				)] for i in range(self.ROW				)]
       lowercase :       Optional[Any]       =			0
       for i in range(self.ROW				):
            for j in range(self.COL				):
                 if visited[i][j] is False and self.graph[i][j] == 1:
                      self.diffs(_A				,						_A				,						_A				)
                      count += 1
       return count