MultiSurrogateSpecificCheckpoint.py 3.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  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 MultiSurrogateSpecificCheckpoint(Callback):
  11. """
  12. MultiSurrogateSpecificCheckpoint is used for keep track of sub-surrogate problem indices
  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. population = self._algo._population
  24. # Do nothing is surrogate analyser does not exist
  25. if population is None:
  26. return
  27. currentEvaluation = self._algo.getGlobalEvaluation()
  28. # backup if necessary
  29. if currentEvaluation % self._every == 0:
  30. logging.info(f"Multi surrogate specific analysis checkpoint is done into {self._filepath}")
  31. line = ''
  32. fitness_list = [ s.fitness() for s in population ]
  33. fitness_data = ' '.join(list(map(str, fitness_list)))
  34. for s in population:
  35. s_data = ' '.join(list(map(str, s._data)))
  36. line += s_data + ';'
  37. line += fitness_data
  38. line += '\n'
  39. # check if file exists
  40. if not os.path.exists(self._filepath):
  41. with open(self._filepath, 'w') as f:
  42. f.write(line)
  43. else:
  44. with open(self._filepath, 'a') as f:
  45. f.write(line)
  46. def load(self):
  47. """
  48. Load previous population
  49. """
  50. if os.path.exists(self._filepath):
  51. logging.info('Load population solutions from last checkpoint')
  52. with open(self._filepath) as f:
  53. # get last line and read data
  54. lastline = f.readlines()[-1].replace('\n', '')
  55. data = lastline.split(';')
  56. fitness_scores = list(map(float, data[-1].split(' ')))
  57. for i, solution_data in enumerate(data[:-1]):
  58. self._algo._population[i]._data = list(map(int, solution_data.split(' ')))
  59. self._algo._population[i]._score = fitness_scores[i]
  60. print(macop_line())
  61. print(macop_text(f' MultiSurrogateSpecificCheckpoint found from `{self._filepath}` file. Start running using previous `population` values'))
  62. for i, s in enumerate(self._algo._population):
  63. print(f'Population[{i}]: best solution fitness is {s.fitness()}')
  64. else:
  65. print(macop_text('No backup found... Start running using new `population` values'))
  66. logging.info("Can't load MultiSurrogateSpecific backup... Backup filepath not valid in MultiSurrogateCheckpoint")
  67. print(macop_line())