

'''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      lowercase_      (lowerCamelCase__ ):




  """simple docstring"""




  def __init__( self  :	List[str]     ):
   __lowercase				    =					[]






  def    SCREAMING_SNAKE_CASE ( self  :	Optional[int]   ,lowercase__  :	List[str]   ,lowercase__  :	Tuple   ,lowercase__  :	str   ,**lowercase__  :	Any     ):
   self.events.append('''on_init_end'''     )






  def    SCREAMING_SNAKE_CASE ( self  :	Any   ,lowercase__  :	List[str]   ,lowercase__  :	Optional[Any]   ,lowercase__  :	int   ,**lowercase__  :	Optional[int]     ):
   self.events.append('''on_train_begin'''     )






  def    SCREAMING_SNAKE_CASE ( self  :	Tuple   ,lowercase__  :	Tuple   ,lowercase__  :	int   ,lowercase__  :	int   ,**lowercase__  :	List[str]     ):
   self.events.append('''on_train_end'''     )






  def    SCREAMING_SNAKE_CASE ( self  :	str   ,lowercase__  :	Any   ,lowercase__  :	Union[str, Any]   ,lowercase__  :	Any   ,**lowercase__  :	Optional[Any]     ):
   self.events.append('''on_epoch_begin'''     )






  def    SCREAMING_SNAKE_CASE ( self  :	Tuple   ,lowercase__  :	Optional[Any]   ,lowercase__  :	int   ,lowercase__  :	Any   ,**lowercase__  :	Optional[int]     ):
   self.events.append('''on_epoch_end'''     )






  def    SCREAMING_SNAKE_CASE ( self  :	int   ,lowercase__  :	List[str]   ,lowercase__  :	str   ,lowercase__  :	List[str]   ,**lowercase__  :	List[str]     ):
   self.events.append('''on_step_begin'''     )






  def    SCREAMING_SNAKE_CASE ( self  :	int   ,lowercase__  :	Union[str, Any]   ,lowercase__  :	int   ,lowercase__  :	Optional[int]   ,**lowercase__  :	Dict     ):
   self.events.append('''on_step_end'''     )






  def    SCREAMING_SNAKE_CASE ( self  :	str   ,lowercase__  :	Any   ,lowercase__  :	Tuple   ,lowercase__  :	Union[str, Any]   ,**lowercase__  :	Any     ):
   self.events.append('''on_evaluate'''     )






  def    SCREAMING_SNAKE_CASE ( self  :	Optional[int]   ,lowercase__  :	str   ,lowercase__  :	Union[str, Any]   ,lowercase__  :	int   ,**lowercase__  :	Optional[Any]     ):
   self.events.append('''on_predict'''     )






  def    SCREAMING_SNAKE_CASE ( self  :	Optional[Any]   ,lowercase__  :	List[str]   ,lowercase__  :	Union[str, Any]   ,lowercase__  :	Optional[Any]   ,**lowercase__  :	int     ):
   self.events.append('''on_save'''     )






  def    SCREAMING_SNAKE_CASE ( self  :	Dict   ,lowercase__  :	List[str]   ,lowercase__  :	Tuple   ,lowercase__  :	List[str]   ,**lowercase__  :	List[str]     ):
   self.events.append('''on_log'''     )






  def    SCREAMING_SNAKE_CASE ( self  :	int   ,lowercase__  :	str   ,lowercase__  :	int   ,lowercase__  :	Dict   ,**lowercase__  :	str     ):
   self.events.append('''on_prediction_step'''     )






@require_torch
class      lowercase_      (unittest.TestCase ):




  """simple docstring"""




  def    SCREAMING_SNAKE_CASE ( self  :	List[str]     ):
   __lowercase				    =					tempfile.mkdtemp()






  def    SCREAMING_SNAKE_CASE ( self  :	Union[str, Any]     ):
   shutil.rmtree(self.output_dir     )






  def    SCREAMING_SNAKE_CASE ( self  :	List[Any]   ,lowercase__  :	Optional[Any]=0   ,lowercase__  :	Any=0   ,lowercase__  :	Tuple=6_4   ,lowercase__  :	Optional[int]=6_4   ,lowercase__  :	Optional[Any]=None   ,lowercase__  :	str=False   ,**lowercase__  :	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.
   __lowercase				    =					RegressionDataset(length=lowercase__     )
   __lowercase				    =					RegressionDataset(length=lowercase__     )
   __lowercase				    =					RegressionModelConfig(a=lowercase__   ,b=lowercase__     )
   __lowercase				    =					RegressionPreTrainedModel(lowercase__     )

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






  def    SCREAMING_SNAKE_CASE ( self  :	List[Any]   ,lowercase__  :	Optional[int]   ,lowercase__  :	Any     ):
   self.assertEqual(len(lowercase__     )   ,len(lowercase__     )     )

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

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






  def    SCREAMING_SNAKE_CASE ( self  :	List[Any]     ):
   __lowercase				    =					DEFAULT_CALLBACKS.copy() + [ProgressCallback]
   __lowercase				    =					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__     )

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

   __lowercase				    =					self.get_trainer()
   __lowercase				    =					trainer.callback_handler.callbacks[0]
   __lowercase				    =					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    SCREAMING_SNAKE_CASE ( self  :	Dict     ):
   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__     )

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

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

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

   __lowercase				    =					self.get_trainer(callbacks=[MyTestTrainerCallback]   ,eval_steps=5   ,evaluation_strategy='''steps'''     )
   trainer.train()
   __lowercase				    =					trainer.callback_handler.callbacks[-2].events
   self.assertEqual(lowercase__   ,self.get_expected_events(lowercase__     )     )

   __lowercase				    =					self.get_trainer(callbacks=[MyTestTrainerCallback]   ,evaluation_strategy='''epoch'''     )
   trainer.train()
   __lowercase				    =					trainer.callback_handler.callbacks[-2].events
   self.assertEqual(lowercase__   ,self.get_expected_events(lowercase__     )     )

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






'''simple docstring'''

# Author: OMKAR PATHAK, Nwachukwu Chidiebere

# Use a Python dictionary to construct the graph.
from __future__ import annotations

from pprint import pformat
from typing import Generic, TypeVar

__lowerCAmelCase		     =  TypeVar('''T''')

class       __magic_name__		(       Generic[T]  ):







							def __init__(				self   :			int      ,_UpperCAmelCase   :			bool = True						):
									_a	: dict[T, list[T]]								=				{}  # dictionary of lists
									_a	: Tuple								=				directed







							def 				__lowercase     (				self   :			Union[str, Any]      ,_UpperCAmelCase   :			T      ,_UpperCAmelCase   :			T						):

									if not self.directed:  # For undirected graphs
											# if both source vertex and destination vertex are both present in the
											# adjacency list, add destination vertex to source vertex list of adjacent
											# vertices and add source vertex to destination vertex list of adjacent
											# vertices.
											if source_vertex in self.adj_list and destination_vertex in self.adj_list:
													self.adj_list[source_vertex].append(_UpperCAmelCase						)
													self.adj_list[destination_vertex].append(_UpperCAmelCase						)
											# if only source vertex is present in adjacency list, add destination vertex
											# to source vertex list of adjacent vertices, then create a new vertex with
											# destination vertex as key and assign a list containing the source vertex
											# as it's first adjacent vertex.
											elif source_vertex in self.adj_list:
													self.adj_list[source_vertex].append(_UpperCAmelCase						)
													_a	: Optional[Any]								=				[source_vertex]
											# if only destination vertex is present in adjacency list, add source vertex
											# to destination vertex list of adjacent vertices, then create a new vertex
											# with source vertex as key and assign a list containing the source vertex
											# as it's first adjacent vertex.
											elif destination_vertex in self.adj_list:
													self.adj_list[destination_vertex].append(_UpperCAmelCase						)
													_a	: Union[str, Any]								=				[destination_vertex]
											# if both source vertex and destination vertex are not present in adjacency
											# list, create a new vertex with source vertex as key and assign a list
											# containing the destination vertex as it's first adjacent vertex also
											# create a new vertex with destination vertex as key and assign a list
											# containing the source vertex as it's first adjacent vertex.
											else:
													_a	: Union[str, Any]								=				[destination_vertex]
													_a	: Optional[int]								=				[source_vertex]
									else:  # For directed graphs
											# if both source vertex and destination vertex are present in adjacency
											# list, add destination vertex to source vertex list of adjacent vertices.
											if source_vertex in self.adj_list and destination_vertex in self.adj_list:
													self.adj_list[source_vertex].append(_UpperCAmelCase						)
											# if only source vertex is present in adjacency list, add destination
											# vertex to source vertex list of adjacent vertices and create a new vertex
											# with destination vertex as key, which has no adjacent vertex
											elif source_vertex in self.adj_list:
													self.adj_list[source_vertex].append(_UpperCAmelCase						)
													_a	: int								=				[]
											# if only destination vertex is present in adjacency list, create a new
											# vertex with source vertex as key and assign a list containing destination
											# vertex as first adjacent vertex
											elif destination_vertex in self.adj_list:
													_a	: Tuple								=				[destination_vertex]
											# if both source vertex and destination vertex are not present in adjacency
											# list, create a new vertex with source vertex as key and a list containing
											# destination vertex as it's first adjacent vertex. Then create a new vertex
											# with destination vertex as key, which has no adjacent vertex
											else:
													_a	: Tuple								=				[destination_vertex]
													_a	: str								=				[]

									return self







							def __repr__(				self   :			int						):
									return pformat(self.adj_list						)




