


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 					__SCREAMING_SNAKE_CASE (      snake_case__ ):


      def __init__(				self		):
            lowercase      : Any        =		[]


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    **SCREAMING_SNAKE_CASE__		):
            self.events.append('''on_init_end'''		)


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    **SCREAMING_SNAKE_CASE__		):
            self.events.append('''on_train_begin'''		)


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    **SCREAMING_SNAKE_CASE__		):
            self.events.append('''on_train_end'''		)


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    **SCREAMING_SNAKE_CASE__		):
            self.events.append('''on_epoch_begin'''		)


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    **SCREAMING_SNAKE_CASE__		):
            self.events.append('''on_epoch_end'''		)


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    **SCREAMING_SNAKE_CASE__		):
            self.events.append('''on_step_begin'''		)


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    **SCREAMING_SNAKE_CASE__		):
            self.events.append('''on_step_end'''		)


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    **SCREAMING_SNAKE_CASE__		):
            self.events.append('''on_evaluate'''		)


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    **SCREAMING_SNAKE_CASE__		):
            self.events.append('''on_predict'''		)


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    **SCREAMING_SNAKE_CASE__		):
            self.events.append('''on_save'''		)


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    **SCREAMING_SNAKE_CASE__		):
            self.events.append('''on_log'''		)






      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__				,    **SCREAMING_SNAKE_CASE__		):
            self.events.append('''on_prediction_step'''		)







@require_torch
class 					__SCREAMING_SNAKE_CASE (      unittest.TestCase ):


      def 							__lowerCamelCase      (				self		):
            lowercase      : List[str]        =		tempfile.mkdtemp()


      def 							__lowerCamelCase      (				self		):
            shutil.rmtree(self.output_dir		)


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__=0				,    SCREAMING_SNAKE_CASE__=0				,    SCREAMING_SNAKE_CASE__=64				,    SCREAMING_SNAKE_CASE__=64				,    SCREAMING_SNAKE_CASE__=None				,    SCREAMING_SNAKE_CASE__=False				,    **SCREAMING_SNAKE_CASE__		):
            # 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.
            lowercase      : Optional[Any]        =		RegressionDataset(length=_UpperCAmelCase		)
            lowercase      : List[Any]        =		RegressionDataset(length=_UpperCAmelCase		)
            lowercase      : Union[str, Any]        =		RegressionModelConfig(a=_UpperCAmelCase				,    b=_UpperCAmelCase		)
            lowercase      : List[Any]        =		RegressionPreTrainedModel(_UpperCAmelCase		)

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


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__				,    SCREAMING_SNAKE_CASE__		):
            self.assertEqual(len(_UpperCAmelCase		)				,    len(_UpperCAmelCase		)		)

            # Order doesn't matter
            lowercase      : Optional[Any]        =		sorted(_UpperCAmelCase				,    key=lambda SCREAMING_SNAKE_CASE__		: cb.__name__ if isinstance(_UpperCAmelCase				,    _UpperCAmelCase		) else cb.__class__.__name__		)
            lowercase      : Optional[int]        =		sorted(_UpperCAmelCase				,    key=lambda SCREAMING_SNAKE_CASE__		: cb.__name__ if isinstance(_UpperCAmelCase				,    _UpperCAmelCase		) else cb.__class__.__name__		)

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


      def 							__lowerCamelCase      (				self				,    SCREAMING_SNAKE_CASE__		):
            lowercase      : Optional[Any]        =		['''on_init_end''', '''on_train_begin''']
            lowercase      : Union[str, Any]        =		0
            lowercase      : Any        =		len(trainer.get_eval_dataloader()		)
            lowercase      : List[Any]        =		['''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(_UpperCAmelCase		):
                        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 							__lowerCamelCase      (				self		):
            lowercase      : Dict        =		self.get_trainer()
            lowercase      : List[Any]        =		DEFAULT_CALLBACKS.copy() + [ProgressCallback]
            self.check_callbacks_equality(trainer.callback_handler.callbacks				,    _UpperCAmelCase		)

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

            # TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
            lowercase      : str        =		self.get_trainer(disable_tqdm=_UpperCAmelCase		)
            lowercase      : List[Any]        =		DEFAULT_CALLBACKS.copy() + [PrinterCallback]
            self.check_callbacks_equality(trainer.callback_handler.callbacks				,    _UpperCAmelCase		)


      def 							__lowerCamelCase      (				self		):
            lowercase      : int        =		DEFAULT_CALLBACKS.copy() + [ProgressCallback]
            lowercase      : Optional[int]        =		self.get_trainer()

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

            lowercase      : Optional[Any]        =		self.get_trainer()
            lowercase      : str        =		trainer.pop_callback(_UpperCAmelCase		)
            self.assertEqual(cb.__class__				,    _UpperCAmelCase		)
            self.check_callbacks_equality(trainer.callback_handler.callbacks				,    _UpperCAmelCase		)

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

            # We can also add, pop, or remove by instance
            lowercase      : Optional[int]        =		self.get_trainer()
            lowercase      : Dict        =		trainer.callback_handler.callbacks[0]
            trainer.remove_callback(_UpperCAmelCase		)
            expected_callbacks.remove(_UpperCAmelCase		)
            self.check_callbacks_equality(trainer.callback_handler.callbacks				,    _UpperCAmelCase		)

            lowercase      : Any        =		self.get_trainer()
            lowercase      : Tuple        =		trainer.callback_handler.callbacks[0]
            lowercase      : Optional[Any]        =		trainer.pop_callback(_UpperCAmelCase		)
            self.assertEqual(_UpperCAmelCase				,    _UpperCAmelCase		)
            self.check_callbacks_equality(trainer.callback_handler.callbacks				,    _UpperCAmelCase		)

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






      def 							__lowerCamelCase      (				self		):
            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=_UpperCAmelCase		)

            lowercase      : Union[str, Any]        =		self.get_trainer(callbacks=[MyTestTrainerCallback]		)
            trainer.train()
            lowercase      : str        =		trainer.callback_handler.callbacks[-2].events
            self.assertEqual(_UpperCAmelCase				,    self.get_expected_events(_UpperCAmelCase		)		)

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

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

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

            lowercase      : Tuple        =		self.get_trainer(callbacks=[MyTestTrainerCallback]				,    evaluation_strategy='''epoch'''		)
            trainer.train()
            lowercase      : Union[str, Any]        =		trainer.callback_handler.callbacks[-2].events
            self.assertEqual(_UpperCAmelCase				,    self.get_expected_events(_UpperCAmelCase		)		)

            # A bit of everything
            lowercase      : List[str]        =		self.get_trainer(
                callbacks=[MyTestTrainerCallback]				,    logging_steps=3				,    save_steps=10				,    eval_steps=5				,    evaluation_strategy='''steps'''				,    )
            trainer.train()
            lowercase      : int        =		trainer.callback_handler.callbacks[-2].events
            self.assertEqual(_UpperCAmelCase				,    self.get_expected_events(_UpperCAmelCase		)		)

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






'''simple docstring'''

from typing import TYPE_CHECKING

from ...utils import _LazyModule


SCREAMING_SNAKE_CASE__     :						Dict	=  {'''tokenization_wav2vec2_phoneme''': ['''Wav2Vec2PhonemeCTCTokenizer''']}


if TYPE_CHECKING:
			from .tokenization_wavaveca_phoneme import WavaVecaPhonemeCTCTokenizer
else:
			import sys

			SCREAMING_SNAKE_CASE__     :						Dict	=  _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)




