SurrogateMonoCheckpoint.py 2.9 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091
  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 SurrogateMonoCheckpoint(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. line = str(currentEvaluation) + ';' + str(surrogate_analyser._n_local_search) + ';' + str(surrogate_analyser._every_ls) + ';' + str(surrogate_analyser._time) + ';' + str(surrogate_analyser._r2) \
  40. + ';' + str(surrogate_analyser._mae) \
  41. + ';' + solutionData + ';' + str(solution.fitness) + ';\n'
  42. # check if file exists
  43. if not os.path.exists(self._filepath):
  44. with open(self._filepath, 'w') as f:
  45. f.write(line)
  46. else:
  47. with open(self._filepath, 'a') as f:
  48. f.write(line)
  49. def load(self):
  50. """
  51. only load global n local search
  52. """
  53. if os.path.exists(self._filepath):
  54. logging.info('Load n local search')
  55. with open(self._filepath) as f:
  56. # get last line and read data
  57. lastline = f.readlines()[-1].replace(';\n', '')
  58. data = lastline.split(';')
  59. n_local_search = int(data[1])
  60. # set k_indices into main algorithm
  61. self._algo._total_n_local_search = n_local_search
  62. print(macop_line())
  63. print(macop_text(f'SurrogateMonoCheckpoint found from `{self._filepath}` file.'))
  64. else:
  65. print(macop_text('No backup found...'))
  66. logging.info("Can't load Surrogate backup... Backup filepath not valid in SurrogateCheckpoint")
  67. print(macop_line())