




'''simple docstring'''



UpperCamelCase_    	=				[
    "DownloadConfig",
    "DownloadManager",
    "DownloadMode",
    "StreamingDownloadManager",
]

from .download_config import DownloadConfig
from .download_manager import DownloadManager, DownloadMode
from .streaming_download_manager import StreamingDownloadManager






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    (  UpperCamelCase_	):






      def __init__(   self		:	str					)   ->		Union[str, Any]:
             """simple docstring"""




             _lowerCamelCase			:						Optional[Any]            =[]






      def 			lowerCamelCase			(   self		:	Dict     ,  lowercase_		:	List[Any]     ,  lowercase_		:	Optional[Any]     ,  lowercase_		:	Optional[int]     ,  **lowercase_		:	List[Any]					)   ->		List[Any]:
             """simple docstring"""




             self.events.append('on_init_end'					)






      def 			lowerCamelCase			(   self		:	int     ,  lowercase_		:	Tuple     ,  lowercase_		:	Any     ,  lowercase_		:	Optional[Any]     ,  **lowercase_		:	Any					)   ->		Union[str, Any]:
             """simple docstring"""




             self.events.append('on_train_begin'					)






      def 			lowerCamelCase			(   self		:	Optional[int]     ,  lowercase_		:	Dict     ,  lowercase_		:	Dict     ,  lowercase_		:	Tuple     ,  **lowercase_		:	str					)   ->		Optional[Any]:
             """simple docstring"""




             self.events.append('on_train_end'					)






      def 			lowerCamelCase			(   self		:	Optional[Any]     ,  lowercase_		:	str     ,  lowercase_		:	Dict     ,  lowercase_		:	int     ,  **lowercase_		:	List[str]					)   ->		Dict:
             """simple docstring"""




             self.events.append('on_epoch_begin'					)






      def 			lowerCamelCase			(   self		:	Dict     ,  lowercase_		:	Any     ,  lowercase_		:	Dict     ,  lowercase_		:	List[Any]     ,  **lowercase_		:	Dict					)   ->		Optional[Any]:
             """simple docstring"""




             self.events.append('on_epoch_end'					)






      def 			lowerCamelCase			(   self		:	List[str]     ,  lowercase_		:	List[Any]     ,  lowercase_		:	Optional[Any]     ,  lowercase_		:	Any     ,  **lowercase_		:	Dict					)   ->		Tuple:
             """simple docstring"""




             self.events.append('on_step_begin'					)






      def 			lowerCamelCase			(   self		:	List[str]     ,  lowercase_		:	Dict     ,  lowercase_		:	str     ,  lowercase_		:	Optional[Any]     ,  **lowercase_		:	Optional[int]					)   ->		Tuple:
             """simple docstring"""




             self.events.append('on_step_end'					)






      def 			lowerCamelCase			(   self		:	Tuple     ,  lowercase_		:	int     ,  lowercase_		:	List[Any]     ,  lowercase_		:	Union[str, Any]     ,  **lowercase_		:	List[Any]					)   ->		Any:
             """simple docstring"""




             self.events.append('on_evaluate'					)






      def 			lowerCamelCase			(   self		:	Tuple     ,  lowercase_		:	Union[str, Any]     ,  lowercase_		:	Dict     ,  lowercase_		:	List[str]     ,  **lowercase_		:	Dict					)   ->		Dict:
             """simple docstring"""




             self.events.append('on_predict'					)






      def 			lowerCamelCase			(   self		:	Union[str, Any]     ,  lowercase_		:	List[str]     ,  lowercase_		:	Optional[int]     ,  lowercase_		:	List[Any]     ,  **lowercase_		:	Optional[int]					)   ->		Any:
             """simple docstring"""




             self.events.append('on_save'					)






      def 			lowerCamelCase			(   self		:	str     ,  lowercase_		:	List[Any]     ,  lowercase_		:	Any     ,  lowercase_		:	Optional[int]     ,  **lowercase_		:	Optional[int]					)   ->		List[str]:
             """simple docstring"""




             self.events.append('on_log'					)






      def 			lowerCamelCase			(   self		:	Any     ,  lowercase_		:	List[Any]     ,  lowercase_		:	int     ,  lowercase_		:	List[Any]     ,  **lowercase_		:	str					)   ->		List[Any]:
             """simple docstring"""




             self.events.append('on_prediction_step'					)




@require_torch
class A    (  unittest.TestCase	):






      def 			lowerCamelCase			(   self		:	List[Any]					)   ->		Union[str, Any]:
             """simple docstring"""




             _lowerCamelCase			:						Dict            =tempfile.mkdtemp()






      def 			lowerCamelCase			(   self		:	Union[str, Any]					)   ->		int:
             """simple docstring"""




             shutil.rmtree(self.output_dir					)






      def 			lowerCamelCase			(   self		:	Optional[Any]     ,  lowercase_		:	Optional[int]=0     ,  lowercase_		:	int=0     ,  lowercase_		:	Dict=64     ,  lowercase_		:	List[str]=64     ,  lowercase_		:	int=None     ,  lowercase_		:	Tuple=False     ,  **lowercase_		:	Dict					)   ->		Optional[int]:
             """simple docstring"""




             _lowerCamelCase			:						int            =RegressionDataset(length=lowercase_					)
             _lowerCamelCase			:						Optional[Any]            =RegressionDataset(length=lowercase_					)
             _lowerCamelCase			:						str            =RegressionModelConfig(a=lowercase_     ,  b=lowercase_					)
             _lowerCamelCase			:						Optional[Any]            =RegressionPreTrainedModel(lowercase_					)

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






      def 			lowerCamelCase			(   self		:	List[str]     ,  lowercase_		:	int     ,  lowercase_		:	int					)   ->		str:
             """simple docstring"""




             self.assertEqual(len(lowercase_					)     ,  len(lowercase_					)					)

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

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






      def 			lowerCamelCase			(   self		:	Dict     ,  lowercase_		:	List[str]					)   ->		str:
             """simple docstring"""




             _lowerCamelCase			:						List[Any]            =['on_init_end', 'on_train_begin']
             _lowerCamelCase			:						Dict            =0
             _lowerCamelCase			:						str            =len(trainer.get_eval_dataloader()					)
             _lowerCamelCase			:						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(lowercase_					):
                           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		:	Optional[Any]					)   ->		List[str]:
             """simple docstring"""




             _lowerCamelCase			:						Any            =self.get_trainer()
             _lowerCamelCase			:						Optional[Any]            =DEFAULT_CALLBACKS.copy() + [ProgressCallback]
             self.check_callbacks_equality(trainer.callback_handler.callbacks     ,  lowercase_					)

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

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






      def 			lowerCamelCase			(   self		:	str					)   ->		Optional[Any]:
             """simple docstring"""




             _lowerCamelCase			:						int            =DEFAULT_CALLBACKS.copy() + [ProgressCallback]
             _lowerCamelCase			:						List[str]            =self.get_trainer()

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

             _lowerCamelCase			:						Tuple            =self.get_trainer()
             _lowerCamelCase			:						Dict            =trainer.pop_callback(lowercase_					)
             self.assertEqual(cb.__class__     ,  lowercase_					)
             self.check_callbacks_equality(trainer.callback_handler.callbacks     ,  lowercase_					)

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

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

             _lowerCamelCase			:						Any            =self.get_trainer()
             _lowerCamelCase			:						Optional[int]            =trainer.callback_handler.callbacks[0]
             _lowerCamelCase			:						Dict            =trainer.pop_callback(lowercase_					)
             self.assertEqual(lowercase_     ,  lowercase_					)
             self.check_callbacks_equality(trainer.callback_handler.callbacks     ,  lowercase_					)

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






      def 			lowerCamelCase			(   self		:	Tuple					)   ->		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=lowercase_					)

             _lowerCamelCase			:						str            =self.get_trainer(callbacks=[MyTestTrainerCallback]					)
             trainer.train()
             _lowerCamelCase			:						List[Any]            =trainer.callback_handler.callbacks[-2].events
             self.assertEqual(lowercase_     ,  self.get_expected_events(lowercase_					)					)

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

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

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

             _lowerCamelCase			:						Optional[int]            =self.get_trainer(callbacks=[MyTestTrainerCallback]     ,  evaluation_strategy='epoch'					)
             trainer.train()
             _lowerCamelCase			:						Optional[int]            =trainer.callback_handler.callbacks[-2].events
             self.assertEqual(lowercase_     ,  self.get_expected_events(lowercase_					)					)

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

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


