

'''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    a__  (							lowerCamelCase_  ):


   def __init__( self  ):





      """simple docstring"""





      _lowercase			:      List[str]			     =   []


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase			,				**_UpperCamelCase  ):





      """simple docstring"""





      self.events.append("on_init_end"  )


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase			,				**_UpperCamelCase  ):





      """simple docstring"""





      self.events.append("on_train_begin"  )


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase			,				**_UpperCamelCase  ):





      """simple docstring"""





      self.events.append("on_train_end"  )


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase			,				**_UpperCamelCase  ):





      """simple docstring"""





      self.events.append("on_epoch_begin"  )


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase			,				**_UpperCamelCase  ):





      """simple docstring"""





      self.events.append("on_epoch_end"  )


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase			,				**_UpperCamelCase  ):





      """simple docstring"""





      self.events.append("on_step_begin"  )


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase			,				**_UpperCamelCase  ):





      """simple docstring"""





      self.events.append("on_step_end"  )


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase			,				**_UpperCamelCase  ):





      """simple docstring"""





      self.events.append("on_evaluate"  )


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase			,				**_UpperCamelCase  ):





      """simple docstring"""





      self.events.append("on_predict"  )


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase			,				**_UpperCamelCase  ):





      """simple docstring"""





      self.events.append("on_save"  )


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase			,				**_UpperCamelCase  ):





      """simple docstring"""





      self.events.append("on_log"  )






   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase			,				**_UpperCamelCase  ):





      """simple docstring"""





      self.events.append("on_prediction_step"  )




@require_torch
class    a__  (							unittest.TestCase  ):


   def 						_lowerCamelCase   ( self  ):





      """simple docstring"""





      _lowercase			:      int			     =   tempfile.mkdtemp()


   def 						_lowerCamelCase   ( self  ):





      """simple docstring"""





      shutil.rmtree(self.output_dir  )


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase=0			,				_UpperCamelCase=0			,				_UpperCamelCase=64			,				_UpperCamelCase=64			,				_UpperCamelCase=None			,				_UpperCamelCase=False			,				**_UpperCamelCase  ):





      """simple docstring"""





      _lowercase			:      Optional[int]			     =   RegressionDataset(length=__a  )
      _lowercase			:      Union[str, Any]			     =   RegressionDataset(length=__a  )
      _lowercase			:      List[str]			     =   RegressionModelConfig(a=__a			,				b=__a  )
      _lowercase			:      str			     =   RegressionPreTrainedModel(__a  )

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


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase  ):





      """simple docstring"""





      self.assertEqual(len(__a  )			,				len(__a  )  )

      # Order doesn't matter
      _lowercase			:      Any			     =   sorted(__a			,				key=lambda _UpperCamelCase  : cb.__name__ if isinstance(__a			,				__a  ) else cb.__class__.__name__  )
      _lowercase			:      str			     =   sorted(__a			,				key=lambda _UpperCamelCase  : cb.__name__ if isinstance(__a			,				__a  ) else cb.__class__.__name__  )

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


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase  ):





      """simple docstring"""





      _lowercase			:      Optional[Any]			     =   ["on_init_end", "on_train_begin"]
      _lowercase			:      str			     =   0
      _lowercase			:      List[Any]			     =   len(trainer.get_eval_dataloader()  )
      _lowercase			:      List[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(__a  ):
            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  ):





      """simple docstring"""





      _lowercase			:      int			     =   self.get_trainer()
      _lowercase			:      str			     =   DEFAULT_CALLBACKS.copy() + [ProgressCallback]
      self.check_callbacks_equality(trainer.callback_handler.callbacks			,				__a  )

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

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


   def 						_lowerCamelCase   ( self  ):





      """simple docstring"""





      _lowercase			:      Dict			     =   DEFAULT_CALLBACKS.copy() + [ProgressCallback]
      _lowercase			:      Dict			     =   self.get_trainer()

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

      _lowercase			:      Tuple			     =   self.get_trainer()
      _lowercase			:      Dict			     =   trainer.pop_callback(__a  )
      self.assertEqual(cb.__class__			,				__a  )
      self.check_callbacks_equality(trainer.callback_handler.callbacks			,				__a  )

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

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

      _lowercase			:      Optional[int]			     =   self.get_trainer()
      _lowercase			:      Optional[Any]			     =   trainer.callback_handler.callbacks[0]
      _lowercase			:      int			     =   trainer.pop_callback(__a  )
      self.assertEqual(__a			,				__a  )
      self.check_callbacks_equality(trainer.callback_handler.callbacks			,				__a  )

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






   def 						_lowerCamelCase   ( self  ):





      """simple docstring"""





      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=__a  )

      _lowercase			:      Tuple			     =   self.get_trainer(callbacks=[MyTestTrainerCallback]  )
      trainer.train()
      _lowercase			:      List[Any]			     =   trainer.callback_handler.callbacks[-2].events
      self.assertEqual(__a			,				self.get_expected_events(__a  )  )

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

      _lowercase			:      List[str]			     =   self.get_trainer(callbacks=[MyTestTrainerCallback]			,				save_steps=5  )
      trainer.train()
      _lowercase			:      List[str]			     =   trainer.callback_handler.callbacks[-2].events
      self.assertEqual(__a			,				self.get_expected_events(__a  )  )

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

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

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

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











import argparse
import copy






def 				A				(					_lowerCamelCase  ):





							'''simple docstring'''




							_lowerCAmelCase     :							int    =    {}

							with open(_lowerCamelCase  ) as f:
														for line in f:
																					if line.split()[0] not in dict_of_neighbours:
																												_lowerCAmelCase     :							Tuple    =    []
																												_list.append([line.split()[1], line.split()[2]]  )
																												_lowerCAmelCase     :							Any    =    _list
																					else:
																												dict_of_neighbours[line.split()[0]].append(
																												    [line.split()[1], line.split()[2]]  )
																					if line.split()[1] not in dict_of_neighbours:
																												_lowerCAmelCase     :							str    =    []
																												_list.append([line.split()[0], line.split()[2]]  )
																												_lowerCAmelCase     :							Any    =    _list
																					else:
																												dict_of_neighbours[line.split()[1]].append(
																												    [line.split()[0], line.split()[2]]  )

							return dict_of_neighbours






def 				A				(					_lowerCamelCase				, _lowerCamelCase  ):





							'''simple docstring'''




							with open(_lowerCamelCase  ) as f:
														_lowerCAmelCase     :							str    =    f.read(1  )
							_lowerCAmelCase     :							str    =    start_node

							_lowerCAmelCase     :							List[str]    =    []

							_lowerCAmelCase     :							Any    =    start_node

							_lowerCAmelCase     :							str    =    0
							while visiting not in first_solution:
														_lowerCAmelCase     :							Dict    =    10_000
														for k in dict_of_neighbours[visiting]:
																					if int(k[1]  ) < int(_lowerCamelCase  ) and k[0] not in first_solution:
																												_lowerCAmelCase     :							List[str]    =    k[1]
																												_lowerCAmelCase     :							List[Any]    =    k[0]

														first_solution.append(_lowerCamelCase  )
														_lowerCAmelCase     :							Optional[int]    =    distance_of_first_solution + int(_lowerCamelCase  )
														_lowerCAmelCase     :							str    =    best_node

							first_solution.append(_lowerCamelCase  )

							_lowerCAmelCase     :							Union[str, Any]    =    0
							for k in dict_of_neighbours[first_solution[-2]]:
														if k[0] == start_node:
																					break
														position += 1

							_lowerCAmelCase     :							Tuple    =    (
							    distance_of_first_solution
							    + int(dict_of_neighbours[first_solution[-2]][position][1]  )
							    - 10_000
							)
							return first_solution, distance_of_first_solution






def 				A				(					_lowerCamelCase				, _lowerCamelCase  ):





							'''simple docstring'''




							_lowerCAmelCase     :							Tuple    =    []

							for n in solution[1:-1]:
														_lowerCAmelCase     :							Dict    =    solution.index(_lowerCamelCase  )
														for kn in solution[1:-1]:
																					_lowerCAmelCase     :							Dict    =    solution.index(_lowerCamelCase  )
																					if n == kn:
																												continue

																					_lowerCAmelCase     :							Optional[int]    =    copy.deepcopy(_lowerCamelCase  )
																					_lowerCAmelCase     :							int    =    kn
																					_lowerCAmelCase     :							Dict    =    n

																					_lowerCAmelCase     :							Optional[int]    =    0

																					for k in _tmp[:-1]:
																												_lowerCAmelCase     :							str    =    _tmp[_tmp.index(_lowerCamelCase  ) + 1]
																												for i in dict_of_neighbours[k]:
																																			if i[0] == next_node:
																																										_lowerCAmelCase     :							Optional[Any]    =    distance + int(i[1]  )
																					_tmp.append(_lowerCamelCase  )

																					if _tmp not in neighborhood_of_solution:
																												neighborhood_of_solution.append(_tmp  )

							_lowerCAmelCase     :							List[Any]    =    len(neighborhood_of_solution[0]  ) - 1

							neighborhood_of_solution.sort(key=lambda _lowerCamelCase  : x[index_of_last_item_in_the_list]  )
							return neighborhood_of_solution






def 				A				(					_lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase  ):





							'''simple docstring'''




							_lowerCAmelCase     :							List[str]    =    1
							_lowerCAmelCase     :							int    =    first_solution
							_lowerCAmelCase     :							Tuple    =    []
							_lowerCAmelCase     :							Tuple    =    distance_of_first_solution
							_lowerCAmelCase     :							Optional[int]    =    solution

							while count <= iters:
														_lowerCAmelCase     :							int    =    find_neighborhood(_lowerCamelCase				, _lowerCamelCase  )
														_lowerCAmelCase     :							Tuple    =    0
														_lowerCAmelCase     :							Dict    =    neighborhood[index_of_best_solution]
														_lowerCAmelCase     :							int    =    len(_lowerCamelCase  ) - 1

														_lowerCAmelCase     :							Union[str, Any]    =    False
														while not found:
																					_lowerCAmelCase     :							Tuple    =    0
																					while i < len(_lowerCamelCase  ):
																												if best_solution[i] != solution[i]:
																																			_lowerCAmelCase     :							str    =    best_solution[i]
																																			_lowerCAmelCase     :							Tuple    =    solution[i]
																																			break
																												_lowerCAmelCase     :							int    =    i + 1

																					if [first_exchange_node, second_exchange_node] not in tabu_list and [
																					    second_exchange_node,
																					    first_exchange_node,
																					] not in tabu_list:
																												tabu_list.append([first_exchange_node, second_exchange_node]  )
																												_lowerCAmelCase     :							Optional[int]    =    True
																												_lowerCAmelCase     :							Optional[Any]    =    best_solution[:-1]
																												_lowerCAmelCase     :							Tuple    =    neighborhood[index_of_best_solution][best_cost_index]
																												if cost < best_cost:
																																			_lowerCAmelCase     :							Union[str, Any]    =    cost
																																			_lowerCAmelCase     :							List[Any]    =    solution
																					else:
																												_lowerCAmelCase     :							Optional[Any]    =    index_of_best_solution + 1
																												_lowerCAmelCase     :							Optional[Any]    =    neighborhood[index_of_best_solution]

														if len(_lowerCamelCase  ) >= size:
																					tabu_list.pop(0  )

														_lowerCAmelCase     :							int    =    count + 1

							return best_solution_ever, best_cost






def 				A				(					_lowerCamelCase=None  ):





							'''simple docstring'''




							_lowerCAmelCase     :							int    =    generate_neighbours(args.File  )

							_lowerCAmelCase   ,    _lowerCAmelCase     :							List[str]    =    generate_first_solution(
							    args.File				, _lowerCamelCase  )

							_lowerCAmelCase   ,    _lowerCAmelCase     :							Any    =    tabu_search(
							    _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, args.Iterations				, args.Size				, )

							print(F"Best solution: {best_sol}, with total distance: {best_cost}."  )


if __name__ == "__main__":
						_snake_case		=  argparse.ArgumentParser(description="Tabu Search")
						parser.add_argument(
						    "-f",
						    "--File",
						    type=str,
						    help="Path to the file containing the data",
						    required=True,
						)
						parser.add_argument(
						    "-i",
						    "--Iterations",
						    type=int,
						    help="How many iterations the algorithm should perform",
						    required=True,
						)
						parser.add_argument(
						    "-s", "--Size", type=int, help="Size of the tabu list", required=True
						)

						# Pass the arguments to main method
						main(parser.parse_args())
