



'''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     UpperCAmelCase__					(      UpperCAmelCase_):



 def __init__(					self     )					->							int:
  __UpperCamelCase			 =					[]



 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase						,		lowercase						,		**lowercase     )					->							Union[str, Any]:
  self.events.append("""on_init_end"""     )



 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase						,		lowercase						,		**lowercase     )					->							List[Any]:
  self.events.append("""on_train_begin"""     )



 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase						,		lowercase						,		**lowercase     )					->							Any:
  self.events.append("""on_train_end"""     )



 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase						,		lowercase						,		**lowercase     )					->							int:
  self.events.append("""on_epoch_begin"""     )



 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase						,		lowercase						,		**lowercase     )					->							int:
  self.events.append("""on_epoch_end"""     )



 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase						,		lowercase						,		**lowercase     )					->							Union[str, Any]:
  self.events.append("""on_step_begin"""     )



 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase						,		lowercase						,		**lowercase     )					->							Dict:
  self.events.append("""on_step_end"""     )



 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase						,		lowercase						,		**lowercase     )					->							Optional[int]:
  self.events.append("""on_evaluate"""     )



 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase						,		lowercase						,		**lowercase     )					->							Union[str, Any]:
  self.events.append("""on_predict"""     )



 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase						,		lowercase						,		**lowercase     )					->							str:
  self.events.append("""on_save"""     )



 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase						,		lowercase						,		**lowercase     )					->							Dict:
  self.events.append("""on_log"""     )




 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase						,		lowercase						,		**lowercase     )					->							List[Any]:
  self.events.append("""on_prediction_step"""     )


@require_torch
class     UpperCAmelCase__					(      unittest.TestCase):



 def 				__lowerCamelCase			(					self     )					->							int:
  __UpperCamelCase			 =					tempfile.mkdtemp()



 def 				__lowerCamelCase			(					self     )					->							Optional[int]:
  shutil.rmtree(self.output_dir     )



 def 				__lowerCamelCase			(					self						,		lowercase=0						,		lowercase=0						,		lowercase=6_4						,		lowercase=6_4						,		lowercase=None						,		lowercase=False						,		**lowercase     )					->							Union[str, 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.
  __UpperCamelCase			 =					RegressionDataset(length=lowercase     )
  __UpperCamelCase			 =					RegressionDataset(length=lowercase     )
  __UpperCamelCase			 =					RegressionModelConfig(a=lowercase						,		b=lowercase     )
  __UpperCamelCase			 =					RegressionPreTrainedModel(lowercase     )

  __UpperCamelCase			 =					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						,		lowercase						,		lowercase     )					->							Optional[int]:
  self.assertEqual(len(lowercase     )						,		len(lowercase     )     )

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

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



 def 				__lowerCamelCase			(					self     )					->							Tuple:
  __UpperCamelCase			 =					DEFAULT_CALLBACKS.copy() + [ProgressCallback]
  __UpperCamelCase			 =					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     )

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

  __UpperCamelCase			 =					self.get_trainer()
  __UpperCamelCase			 =					trainer.callback_handler.callbacks[0]
  __UpperCamelCase			 =					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     )					->							str:
  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     )

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

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

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

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

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

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

a__  :     int     							=     TypeVar('T')
class     UpperCAmelCase__					(      Generic[T]):



 def __init__(					self						,		lowercase = True     )					->							None:
  __UpperCamelCase			 =					{}  # dictionary of lists
  __UpperCamelCase			 =					directed



 def 				__lowerCamelCase			(					self						,		lowercase						,		lowercase     )					->							GraphAdjacencyList[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(lowercase     )
    self.adj_list[destination_vertex].append(lowercase     )
   # 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(lowercase     )
    __UpperCamelCase			 =					[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(lowercase     )
    __UpperCamelCase			 =					[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:
    __UpperCamelCase			 =					[destination_vertex]
    __UpperCamelCase			 =					[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(lowercase     )
   # 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(lowercase     )
    __UpperCamelCase			 =					[]
   # 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:
    __UpperCamelCase			 =					[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:
    __UpperCamelCase			 =					[destination_vertex]
    __UpperCamelCase			 =					[]

  return self




 def __repr__(					self     )					->							str:
  return pformat(self.adj_list     )





