


'''simple docstring'''






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        __snake_case	(       _SCREAMING_SNAKE_CASE):







    """simple docstring"""



    def __init__(       self   :  Optional[Any]		)       ->  Dict:
      lowerCAmelCase_      :				Dict      =       []

    def __lowercase       (       self   :  Union[str, Any]   ,						lowerCamelCase   :  Union[str, Any]   ,						lowerCamelCase   :  Optional[int]   ,						lowerCamelCase   :  Any   ,						**lowerCamelCase   :  Optional[int]		)       ->  Optional[int]:
      self.events.append("""on_init_end"""		)

    def __lowercase       (       self   :  Union[str, Any]   ,						lowerCamelCase   :  List[Any]   ,						lowerCamelCase   :  List[str]   ,						lowerCamelCase   :  str   ,						**lowerCamelCase   :  Dict		)       ->  List[str]:
      self.events.append("""on_train_begin"""		)

    def __lowercase       (       self   :  List[str]   ,						lowerCamelCase   :  Optional[int]   ,						lowerCamelCase   :  int   ,						lowerCamelCase   :  Dict   ,						**lowerCamelCase   :  Tuple		)       ->  List[str]:
      self.events.append("""on_train_end"""		)

    def __lowercase       (       self   :  str   ,						lowerCamelCase   :  List[str]   ,						lowerCamelCase   :  List[Any]   ,						lowerCamelCase   :  str   ,						**lowerCamelCase   :  List[str]		)       ->  Any:
      self.events.append("""on_epoch_begin"""		)

    def __lowercase       (       self   :  Optional[int]   ,						lowerCamelCase   :  Tuple   ,						lowerCamelCase   :  Dict   ,						lowerCamelCase   :  Any   ,						**lowerCamelCase   :  Tuple		)       ->  Optional[Any]:
      self.events.append("""on_epoch_end"""		)

    def __lowercase       (       self   :  Any   ,						lowerCamelCase   :  Dict   ,						lowerCamelCase   :  Tuple   ,						lowerCamelCase   :  str   ,						**lowerCamelCase   :  Optional[int]		)       ->  Optional[Any]:
      self.events.append("""on_step_begin"""		)

    def __lowercase       (       self   :  Any   ,						lowerCamelCase   :  Optional[int]   ,						lowerCamelCase   :  List[Any]   ,						lowerCamelCase   :  str   ,						**lowerCamelCase   :  str		)       ->  Tuple:
      self.events.append("""on_step_end"""		)

    def __lowercase       (       self   :  List[Any]   ,						lowerCamelCase   :  int   ,						lowerCamelCase   :  int   ,						lowerCamelCase   :  Tuple   ,						**lowerCamelCase   :  List[str]		)       ->  Dict:
      self.events.append("""on_evaluate"""		)

    def __lowercase       (       self   :  Optional[Any]   ,						lowerCamelCase   :  Dict   ,						lowerCamelCase   :  List[Any]   ,						lowerCamelCase   :  Dict   ,						**lowerCamelCase   :  Optional[Any]		)       ->  int:
      self.events.append("""on_predict"""		)

    def __lowercase       (       self   :  Optional[int]   ,						lowerCamelCase   :  Any   ,						lowerCamelCase   :  List[str]   ,						lowerCamelCase   :  Any   ,						**lowerCamelCase   :  Optional[Any]		)       ->  int:
      self.events.append("""on_save"""		)

    def __lowercase       (       self   :  Tuple   ,						lowerCamelCase   :  int   ,						lowerCamelCase   :  Tuple   ,						lowerCamelCase   :  int   ,						**lowerCamelCase   :  Optional[Any]		)       ->  Union[str, Any]:
      self.events.append("""on_log"""		)





    def __lowercase       (       self   :  Any   ,						lowerCamelCase   :  List[Any]   ,						lowerCamelCase   :  int   ,						lowerCamelCase   :  Tuple   ,						**lowerCamelCase   :  Union[str, Any]		)       ->  Dict:
      self.events.append("""on_prediction_step"""		)







@require_torch
class        __snake_case	(       unittest.TestCase):







    """simple docstring"""



    def __lowercase       (       self   :  Optional[int]		)       ->  List[str]:
      lowerCAmelCase_      :				Optional[int]      =       tempfile.mkdtemp()

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

    def __lowercase       (       self   :  Tuple   ,						lowerCamelCase   :  Tuple=0   ,						lowerCamelCase   :  List[str]=0   ,						lowerCamelCase   :  Dict=64   ,						lowerCamelCase   :  Dict=64   ,						lowerCamelCase   :  Optional[Any]=None   ,						lowerCamelCase   :  Optional[int]=False   ,						**lowerCamelCase   :  Optional[Any]		)       ->  List[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_      :				Optional[Any]      =       RegressionDataset(length=lowerCamelCase		)
      lowerCAmelCase_      :				Union[str, Any]      =       RegressionDataset(length=lowerCamelCase		)
      lowerCAmelCase_      :				int      =       RegressionModelConfig(a=lowerCamelCase   ,						b=lowerCamelCase		)
      lowerCAmelCase_      :				str      =       RegressionPreTrainedModel(lowerCamelCase		)

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

    def __lowercase       (       self   :  Optional[int]   ,						lowerCamelCase   :  Dict   ,						lowerCamelCase   :  List[str]		)       ->  List[str]:
      self.assertEqual(len(lowerCamelCase		)   ,						len(lowerCamelCase		)		)

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

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

    def __lowercase       (       self   :  Dict   ,						lowerCamelCase   :  Union[str, Any]		)       ->  str:
      lowerCAmelCase_      :				Optional[Any]      =       ["""on_init_end""", """on_train_begin"""]
      lowerCAmelCase_      :				int      =       0
      lowerCAmelCase_      :				str      =       len(trainer.get_eval_dataloader()		)
      lowerCAmelCase_      :				str      =       ["""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(lowerCamelCase		):
          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 __lowercase       (       self   :  List[Any]		)       ->  Any:
      lowerCAmelCase_      :				Dict      =       self.get_trainer()
      lowerCAmelCase_      :				Optional[Any]      =       DEFAULT_CALLBACKS.copy() + [ProgressCallback]
      self.check_callbacks_equality(trainer.callback_handler.callbacks   ,						lowerCamelCase		)

      # Callbacks passed at init are added to the default callbacks
      lowerCAmelCase_      :				Union[str, Any]      =       self.get_trainer(callbacks=[MyTestTrainerCallback]		)
      expected_callbacks.append(lowerCamelCase		)
      self.check_callbacks_equality(trainer.callback_handler.callbacks   ,						lowerCamelCase		)

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

    def __lowercase       (       self   :  Tuple		)       ->  Tuple:
      lowerCAmelCase_      :				List[Any]      =       DEFAULT_CALLBACKS.copy() + [ProgressCallback]
      lowerCAmelCase_      :				str      =       self.get_trainer()

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

      lowerCAmelCase_      :				List[str]      =       self.get_trainer()
      lowerCAmelCase_      :				Any      =       trainer.pop_callback(lowerCamelCase		)
      self.assertEqual(cb.__class__   ,						lowerCamelCase		)
      self.check_callbacks_equality(trainer.callback_handler.callbacks   ,						lowerCamelCase		)

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

      # We can also add, pop, or remove by instance
      lowerCAmelCase_      :				str      =       self.get_trainer()
      lowerCAmelCase_      :				Tuple      =       trainer.callback_handler.callbacks[0]
      trainer.remove_callback(lowerCamelCase		)
      expected_callbacks.remove(lowerCamelCase		)
      self.check_callbacks_equality(trainer.callback_handler.callbacks   ,						lowerCamelCase		)

      lowerCAmelCase_      :				List[Any]      =       self.get_trainer()
      lowerCAmelCase_      :				Tuple      =       trainer.callback_handler.callbacks[0]
      lowerCAmelCase_      :				Tuple      =       trainer.pop_callback(lowerCamelCase		)
      self.assertEqual(lowerCamelCase   ,						lowerCamelCase		)
      self.check_callbacks_equality(trainer.callback_handler.callbacks   ,						lowerCamelCase		)

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





    def __lowercase       (       self   :  str		)       ->  Union[str, 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=lowerCamelCase		)

      lowerCAmelCase_      :				List[Any]      =       self.get_trainer(callbacks=[MyTestTrainerCallback]		)
      trainer.train()
      lowerCAmelCase_      :				Dict      =       trainer.callback_handler.callbacks[-2].events
      self.assertEqual(lowerCamelCase   ,						self.get_expected_events(lowerCamelCase		)		)

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

      lowerCAmelCase_      :				Dict      =       self.get_trainer(callbacks=[MyTestTrainerCallback]   ,						save_steps=5		)
      trainer.train()
      lowerCAmelCase_      :				Union[str, Any]      =       trainer.callback_handler.callbacks[-2].events
      self.assertEqual(lowerCamelCase   ,						self.get_expected_events(lowerCamelCase		)		)

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

      lowerCAmelCase_      :				List[str]      =       self.get_trainer(callbacks=[MyTestTrainerCallback]   ,						evaluation_strategy="""epoch"""		)
      trainer.train()
      lowerCAmelCase_      :				List[Any]      =       trainer.callback_handler.callbacks[-2].events
      self.assertEqual(lowerCamelCase   ,						self.get_expected_events(lowerCamelCase		)		)

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

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







'''simple docstring'''






from ...utils import is_note_seq_available, is_transformers_available, is_torch_available
from ...utils import OptionalDependencyNotAvailable


try:
  if not (is_transformers_available() and is_torch_available()):
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  from ...utils.dummy_torch_and_transformers_objects import *  # noqa F403
else:
  from .notes_encoder import SpectrogramNotesEncoder
  from .continous_encoder import SpectrogramContEncoder
  from .pipeline_spectrogram_diffusion import (
      SpectrogramContEncoder,
      SpectrogramDiffusionPipeline,
      TaFilmDecoder,
  )

try:
  if not (is_transformers_available() and is_torch_available() and is_note_seq_available()):
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  from ...utils.dummy_transformers_and_torch_and_note_seq_objects import *  # noqa F403
else:
  from .midi_utils import MidiProcessor



