
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	)							-> List[Any]:
							lowerCAmelCase_         =      []




				def 	__a		(   self							,							_UpperCamelCase							,							_UpperCamelCase							,							_UpperCamelCase							,							**_UpperCamelCase	)							-> Optional[Any]:
							self.events.append("on_init_end"	)




				def 	__a		(   self							,							_UpperCamelCase							,							_UpperCamelCase							,							_UpperCamelCase							,							**_UpperCamelCase	)							-> Union[str, Any]:
							self.events.append("on_train_begin"	)




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




				def 	__a		(   self							,							_UpperCamelCase							,							_UpperCamelCase							,							_UpperCamelCase							,							**_UpperCamelCase	)							-> str:
							self.events.append("on_epoch_begin"	)




				def 	__a		(   self							,							_UpperCamelCase							,							_UpperCamelCase							,							_UpperCamelCase							,							**_UpperCamelCase	)							-> Optional[int]:
							self.events.append("on_epoch_end"	)




				def 	__a		(   self							,							_UpperCamelCase							,							_UpperCamelCase							,							_UpperCamelCase							,							**_UpperCamelCase	)							-> Tuple:
							self.events.append("on_step_begin"	)




				def 	__a		(   self							,							_UpperCamelCase							,							_UpperCamelCase							,							_UpperCamelCase							,							**_UpperCamelCase	)							-> Union[str, Any]:
							self.events.append("on_step_end"	)




				def 	__a		(   self							,							_UpperCamelCase							,							_UpperCamelCase							,							_UpperCamelCase							,							**_UpperCamelCase	)							-> List[str]:
							self.events.append("on_evaluate"	)




				def 	__a		(   self							,							_UpperCamelCase							,							_UpperCamelCase							,							_UpperCamelCase							,							**_UpperCamelCase	)							-> Any:
							self.events.append("on_predict"	)




				def 	__a		(   self							,							_UpperCamelCase							,							_UpperCamelCase							,							_UpperCamelCase							,							**_UpperCamelCase	)							-> List[str]:
							self.events.append("on_save"	)




				def 	__a		(   self							,							_UpperCamelCase							,							_UpperCamelCase							,							_UpperCamelCase							,							**_UpperCamelCase	)							-> Union[str, Any]:
							self.events.append("on_log"	)




				def 	__a		(   self							,							_UpperCamelCase							,							_UpperCamelCase							,							_UpperCamelCase							,							**_UpperCamelCase	)							-> Tuple:
							self.events.append("on_prediction_step"	)





@require_torch
class       _lowerCAmelCase   (      unittest.TestCase			):




				def 	__a		(   self	)							-> int:
							lowerCAmelCase_         =      tempfile.mkdtemp()




				def 	__a		(   self	)							-> Any:
							shutil.rmtree(self.output_dir	)




				def 	__a		(   self							,							_UpperCamelCase=0							,							_UpperCamelCase=0							,							_UpperCamelCase=64							,							_UpperCamelCase=64							,							_UpperCamelCase=None							,							_UpperCamelCase=False							,							**_UpperCamelCase	)							-> str:
							# 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.
							lowerCAmelCase_         =      RegressionDataset(length=UpperCAmelCase_	)
							lowerCAmelCase_         =      RegressionDataset(length=UpperCAmelCase_	)
							lowerCAmelCase_         =      RegressionModelConfig(a=UpperCAmelCase_							,							b=UpperCAmelCase_	)
							lowerCAmelCase_         =      RegressionPreTrainedModel(UpperCAmelCase_	)

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




				def 	__a		(   self							,							_UpperCamelCase							,							_UpperCamelCase	)							-> Optional[Any]:
							self.assertEqual(len(UpperCAmelCase_	)							,							len(UpperCAmelCase_	)	)

							# Order doesn't matter
							lowerCAmelCase_         =      sorted(UpperCAmelCase_							,							key=lambda _UpperCamelCase	: cb.__name__ if isinstance(UpperCAmelCase_							,							UpperCAmelCase_	) else cb.__class__.__name__	)
							lowerCAmelCase_         =      sorted(UpperCAmelCase_							,							key=lambda _UpperCamelCase	: 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 	__a		(   self							,							_UpperCamelCase	)							-> Any:
							lowerCAmelCase_         =      ["on_init_end", "on_train_begin"]
							lowerCAmelCase_         =      0
							lowerCAmelCase_         =      len(trainer.get_eval_dataloader()	)
							lowerCAmelCase_         =      ["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 	__a		(   self	)							-> Union[str, Any]:
							lowerCAmelCase_         =      self.get_trainer()
							lowerCAmelCase_         =      DEFAULT_CALLBACKS.copy() + [ProgressCallback]
							self.check_callbacks_equality(trainer.callback_handler.callbacks							,							UpperCAmelCase_	)

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




				def 	__a		(   self	)							-> List[str]:
							lowerCAmelCase_         =      DEFAULT_CALLBACKS.copy() + [ProgressCallback]
							lowerCAmelCase_         =      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_	)

							lowerCAmelCase_         =      self.get_trainer()
							lowerCAmelCase_         =      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
							lowerCAmelCase_         =      self.get_trainer()
							lowerCAmelCase_         =      trainer.callback_handler.callbacks[0]
							trainer.remove_callback(UpperCAmelCase_	)
							expected_callbacks.remove(UpperCAmelCase_	)
							self.check_callbacks_equality(trainer.callback_handler.callbacks							,							UpperCAmelCase_	)

							lowerCAmelCase_         =      self.get_trainer()
							lowerCAmelCase_         =      trainer.callback_handler.callbacks[0]
							lowerCAmelCase_         =      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 	__a		(   self	)							-> 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=UpperCAmelCase_	)

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

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

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

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

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

							# A bit of everything
							lowerCAmelCase_         =      self.get_trainer(
							    callbacks=[MyTestTrainerCallback]							,							logging_steps=3							,							save_steps=10							,							eval_steps=5							,							evaluation_strategy="steps"							,							)
							trainer.train()
							lowerCAmelCase_         =      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:
										lowerCAmelCase_         =      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 OptionalDependencyNotAvailable, _LazyModule, is_torch_available


SCREAMING_SNAKE_CASE__          =  {
    "configuration_clap": [
        "CLAP_PRETRAINED_MODEL_ARCHIVE_LIST",
        "ClapAudioConfig",
        "ClapConfig",
        "ClapTextConfig",
    ],
    "processing_clap": ["ClapProcessor"],
}

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 SCREAMING_SNAKE_CASE__          =  [
     "CLAP_PRETRAINED_MODEL_ARCHIVE_LIST",
     "ClapModel",
     "ClapPreTrainedModel",
     "ClapTextModel",
     "ClapTextModelWithProjection",
     "ClapAudioModel",
     "ClapAudioModelWithProjection",
 ]
 SCREAMING_SNAKE_CASE__          =  ["ClapFeatureExtractor"]

if TYPE_CHECKING:
 from .configuration_clap import (
     CLAP_PRETRAINED_MODEL_ARCHIVE_LIST,
     ClapAudioConfig,
     ClapConfig,
     ClapTextConfig,
 )
 from .processing_clap import ClapProcessor

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .feature_extraction_clap import ClapFeatureExtractor
  from .modeling_clap import (
      CLAP_PRETRAINED_MODEL_ARCHIVE_LIST,
      ClapAudioModel,
      ClapAudioModelWithProjection,
      ClapModel,
      ClapPreTrainedModel,
      ClapTextModel,
      ClapTextModelWithProjection,
  )


else:
 import sys

 SCREAMING_SNAKE_CASE__          =  _LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)




