SurrogateCheckpoint.py 3.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293
  1. """Basic Checkpoint class implementation
  2. """
  3. # main imports
  4. import os
  5. import logging
  6. import numpy as np
  7. # module imports
  8. from macop.callbacks.Callback import Callback
  9. from macop.utils.color import macop_text, macop_line
  10. class SurrogateCheckpoint(Callback):
  11. """
  12. SurrogateCheckpoint is used for logging training data information about surrogate
  13. Attributes:
  14. algo: {Algorithm} -- main algorithm instance reference
  15. every: {int} -- checkpoint frequency used (based on number of evaluations)
  16. filepath: {str} -- file path where checkpoints will be saved
  17. """
  18. def run(self):
  19. """
  20. Check if necessary to do backup based on `every` variable
  21. """
  22. # get current best solution
  23. solution = self._algo._bestSolution
  24. surrogate_analyser = self._algo._surrogate_analyser
  25. # Do nothing is surrogate analyser does not exist
  26. if surrogate_analyser is None:
  27. return
  28. currentEvaluation = self._algo.getGlobalEvaluation()
  29. # backup if necessary
  30. if currentEvaluation % self._every == 0:
  31. logging.info(f"Surrogate analysis checkpoint is done into {self._filepath}")
  32. solutionData = ""
  33. solutionSize = len(solution._data)
  34. for index, val in enumerate(solution._data):
  35. solutionData += str(val)
  36. if index < solutionSize - 1:
  37. solutionData += ' '
  38. # get score of r² and mae
  39. r2_data = ' '.join(list(map(str, surrogate_analyser._r2_scores)))
  40. mae_data = ' '.join(list(map(str, surrogate_analyser._mae_scores)))
  41. line = str(currentEvaluation) + ';' + str(surrogate_analyser._n_local_search) + ';' + str(surrogate_analyser._every_ls) + ';' + str(surrogate_analyser._time) + ';' + r2_data + ';' + str(surrogate_analyser._r2) \
  42. + ';' + mae_data + ';' + str(surrogate_analyser._mae) \
  43. + ';' + solutionData + ';' + str(solution.fitness()) + ';\n'
  44. # check if file exists
  45. if not os.path.exists(self._filepath):
  46. with open(self._filepath, 'w') as f:
  47. f.write(line)
  48. else:
  49. with open(self._filepath, 'a') as f:
  50. f.write(line)
  51. def load(self):
  52. """
  53. only load global n local search
  54. """
  55. if os.path.exists(self._filepath):
  56. logging.info('Load n local search')
  57. with open(self._filepath) as f:
  58. # get last line and read data
  59. lastline = f.readlines()[-1].replace(';\n', '')
  60. data = lastline.split(';')
  61. n_local_search = int(data[1])
  62. # set k_indices into main algorithm
  63. self._algo._total_n_local_search = n_local_search
  64. print(macop_line())
  65. print(macop_text(f'SurrogateCheckpoint found from `{self._filepath}` file.'))
  66. else:
  67. print(macop_text('No backup found...'))
  68. logging.info("Can't load Surrogate backup... Backup filepath not valid in SurrogateCheckpoint")
  69. print(macop_line())