

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__				):
						"""simple docstring"""






						def __init__(  self  )			->      Optional[Any]:
									a_		:		str      				=      []





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase  )			->      int:
									self.events.append("""on_init_end"""  )





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase  )			->      Tuple:
									self.events.append("""on_train_begin"""  )





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase  )			->      Union[str, Any]:
									self.events.append("""on_train_end"""  )





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase  )			->      Optional[Any]:
									self.events.append("""on_epoch_begin"""  )





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase  )			->      List[Any]:
									self.events.append("""on_epoch_end"""  )





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase  )			->      int:
									self.events.append("""on_step_begin"""  )





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase  )			->      Any:
									self.events.append("""on_step_end"""  )





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase  )			->      str:
									self.events.append("""on_evaluate"""  )





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase  )			->      Optional[Any]:
									self.events.append("""on_predict"""  )





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase  )			->      Any:
									self.events.append("""on_save"""  )





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase  )			->      Optional[Any]:
									self.events.append("""on_log"""  )





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase  )			->      Any:
									self.events.append("""on_prediction_step"""  )







@require_torch
class     A__(unittest.TestCase				):
						"""simple docstring"""






						def        UpperCamelCase__ (  self  )			->      Tuple:
									a_		:		Tuple      				=      tempfile.mkdtemp()





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





						def        UpperCamelCase__ (  self   ,	_lowercase=0   ,	_lowercase=0   ,	_lowercase=64   ,	_lowercase=64   ,	_lowercase=None   ,	_lowercase=False   ,	**_lowercase  )			->      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.
									a_		:		int      				=      RegressionDataset(length=lowercase_  )
									a_		:		Any      				=      RegressionDataset(length=lowercase_  )
									a_		:		int      				=      RegressionModelConfig(a=lowercase_   ,	b=lowercase_  )
									a_		:		Tuple      				=      RegressionPreTrainedModel(lowercase_  )

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





						def        UpperCamelCase__ (  self   ,	_lowercase   ,	_lowercase  )			->      Any:
									self.assertEqual(len(lowercase_  )   ,	len(lowercase_  )  )

									# Order doesn't matter
									a_		:		Any      				=      sorted(lowercase_   ,	key=lambda _lowercase  : cb.__name__ if isinstance(lowercase_   ,	lowercase_  ) else cb.__class__.__name__  )
									a_		:		List[str]      				=      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        UpperCamelCase__ (  self   ,	_lowercase  )			->      Optional[int]:
									a_		:		Tuple      				=      ['''on_init_end''', '''on_train_begin''']
									a_		:		List[Any]      				=      0
									a_		:		Union[str, Any]      				=      len(trainer.get_eval_dataloader()  )
									a_		:		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        UpperCamelCase__ (  self  )			->      Optional[Any]:
									a_		:		Union[str, Any]      				=      self.get_trainer()
									a_		:		Dict      				=      DEFAULT_CALLBACKS.copy() + [ProgressCallback]
									self.check_callbacks_equality(trainer.callback_handler.callbacks   ,	lowercase_  )

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





						def        UpperCamelCase__ (  self  )			->      Any:
									a_		:		int      				=      DEFAULT_CALLBACKS.copy() + [ProgressCallback]
									a_		:		List[Any]      				=      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_  )

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

									a_		:		List[Any]      				=      self.get_trainer()
									a_		:		Optional[int]      				=      trainer.callback_handler.callbacks[0]
									a_		:		Optional[Any]      				=      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        UpperCamelCase__ (  self  )			->      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=lowercase_  )

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

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

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

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

									a_		:		Union[str, Any]      				=      self.get_trainer(callbacks=[MyTestTrainerCallback]   ,	evaluation_strategy="""epoch"""  )
									trainer.train()
									a_		:		Dict      				=      trainer.callback_handler.callbacks[-2].events
									self.assertEqual(lowercase_   ,	self.get_expected_events(lowercase_  )  )

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







"""simple docstring"""

from . import __version__

# Backward compatibility imports, to make sure all those objects can be found in file_utils
from .utils import (
    CLOUDFRONT_DISTRIB_PREFIX,
    CONFIG_NAME,
    DISABLE_TELEMETRY,
    DUMMY_INPUTS,
    DUMMY_MASK,
    ENV_VARS_TRUE_AND_AUTO_VALUES,
    ENV_VARS_TRUE_VALUES,
    FEATURE_EXTRACTOR_NAME,
    FLAX_WEIGHTS_NAME,
    HF_MODULES_CACHE,
    HUGGINGFACE_CO_PREFIX,
    HUGGINGFACE_CO_RESOLVE_ENDPOINT,
    MODEL_CARD_NAME,
    MULTIPLE_CHOICE_DUMMY_INPUTS,
    PYTORCH_PRETRAINED_BERT_CACHE,
    PYTORCH_TRANSFORMERS_CACHE,
    S3_BUCKET_PREFIX,
    SENTENCEPIECE_UNDERLINE,
    SPIECE_UNDERLINE,
    TF2_WEIGHTS_NAME,
    TF_WEIGHTS_NAME,
    TORCH_FX_REQUIRED_VERSION,
    TRANSFORMERS_CACHE,
    TRANSFORMERS_DYNAMIC_MODULE_NAME,
    USE_JAX,
    USE_TF,
    USE_TORCH,
    WEIGHTS_INDEX_NAME,
    WEIGHTS_NAME,
    ContextManagers,
    DummyObject,
    EntryNotFoundError,
    ExplicitEnum,
    ModelOutput,
    PaddingStrategy,
    PushToHubMixin,
    RepositoryNotFoundError,
    RevisionNotFoundError,
    TensorType,
    _LazyModule,
    add_code_sample_docstrings,
    add_end_docstrings,
    add_start_docstrings,
    add_start_docstrings_to_model_forward,
    cached_property,
    copy_func,
    default_cache_path,
    define_sagemaker_information,
    get_cached_models,
    get_file_from_repo,
    get_full_repo_name,
    get_torch_version,
    has_file,
    http_user_agent,
    is_apex_available,
    is_bsa_available,
    is_coloredlogs_available,
    is_datasets_available,
    is_detectrona_available,
    is_faiss_available,
    is_flax_available,
    is_ftfy_available,
    is_in_notebook,
    is_ipex_available,
    is_librosa_available,
    is_offline_mode,
    is_onnx_available,
    is_pandas_available,
    is_phonemizer_available,
    is_protobuf_available,
    is_psutil_available,
    is_pyanvml_available,
    is_pyctcdecode_available,
    is_pytesseract_available,
    is_pytorch_quantization_available,
    is_rjieba_available,
    is_sagemaker_dp_enabled,
    is_sagemaker_mp_enabled,
    is_scipy_available,
    is_sentencepiece_available,
    is_seqio_available,
    is_sklearn_available,
    is_soundfile_availble,
    is_spacy_available,
    is_speech_available,
    is_tensor,
    is_tensorflow_probability_available,
    is_tfaonnx_available,
    is_tf_available,
    is_timm_available,
    is_tokenizers_available,
    is_torch_available,
    is_torch_bfaa_available,
    is_torch_cuda_available,
    is_torch_fx_available,
    is_torch_fx_proxy,
    is_torch_mps_available,
    is_torch_tfaa_available,
    is_torch_tpu_available,
    is_torchaudio_available,
    is_training_run_on_sagemaker,
    is_vision_available,
    replace_return_docstrings,
    requires_backends,
    to_numpy,
    to_py_obj,
    torch_only_method,
)





