Algorithm.py 10.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290
  1. """Abstract Algorithm class used as basic algorithm implementation with some specific initialization
  2. """
  3. # main imports
  4. import logging
  5. import pkgutil
  6. import sys
  7. from ..utils.color import macop_text, macop_line, macop_progress
  8. # Generic algorithm class
  9. class Algorithm():
  10. """Algorithm class used as basic algorithm
  11. This class enables to manage some common usages of operation research algorithms:
  12. - initialization function of solution
  13. - validator function to check if solution is valid or not (based on some criteria)
  14. - evaluation function to give fitness score to a solution
  15. - operators used in order to update solution during search process
  16. - policy process applied when choosing next operator to apply
  17. - callbacks function in order to do some relative stuff every number of evaluation or reload algorithm state
  18. - parent algorithm associated to this new algorithm instance (hierarchy management)
  19. Attributes:
  20. initalizer: {function} -- basic function strategy to initialize solution
  21. evaluator: {function} -- basic function in order to obtained fitness (mono or multiple objectives)
  22. operators: {[Operator]} -- list of operator to use when launching algorithm
  23. policy: {Policy} -- Policy class implementation strategy to select operators
  24. validator: {function} -- basic function to check if solution is valid or not under some constraints
  25. maximise: {bool} -- specify kind of optimisation problem
  26. currentSolution: {Solution} -- current solution managed for current evaluation comparison
  27. bestSolution: {Solution} -- best solution found so far during running algorithm
  28. callbacks: {[Callback]} -- list of Callback class implementation to do some instructions every number of evaluations and `load` when initializing algorithm
  29. parent: {Algorithm} -- parent algorithm reference in case of inner Algorithm instance (optional)
  30. """
  31. def __init__(self,
  32. initalizer,
  33. evaluator,
  34. operators,
  35. policy,
  36. validator,
  37. maximise=True,
  38. parent=None):
  39. # protected members intialization
  40. self._initializer = initalizer
  41. self._evaluator = evaluator
  42. self._operators = operators
  43. self._policy = policy
  44. self._validator = validator
  45. self._callbacks = []
  46. self._bestSolution = None
  47. self._currentSolution = None
  48. # by default
  49. self._numberOfEvaluations = 0
  50. self._maxEvaluations = 0
  51. # other parameters
  52. self._parent = parent # parent algorithm if it's sub algorithm
  53. #self.maxEvaluations = 0 # by default
  54. self._maximise = maximise
  55. # track reference of algorihtm into operator (keep an eye into best solution)
  56. for operator in self._operators:
  57. if self._parent is not None:
  58. operator.setAlgo(self.getParent())
  59. else:
  60. operator.setAlgo(self)
  61. # also track reference for policy
  62. if self._parent is not None:
  63. self._policy.setAlgo(self.getParent())
  64. else:
  65. self._policy.setAlgo(self)
  66. def addCallback(self, _callback):
  67. """Add new callback to algorithm specifying usefull parameters
  68. Args:
  69. _callback: {Callback} -- specific Callback instance
  70. """
  71. # specify current main algorithm reference for callback
  72. if self._parent is not None:
  73. _callback.setAlgo(self.getParent())
  74. else:
  75. _callback.setAlgo(self)
  76. # set as new
  77. self._callbacks.append(_callback)
  78. def resume(self):
  79. """Resume algorithm using Callback instances
  80. """
  81. # load every callback if many things are necessary to do before running algorithm
  82. for callback in self._callbacks:
  83. callback.load()
  84. def getParent(self):
  85. """Recursively find the main parent algorithm attached of the current algorithm
  86. Returns:
  87. {Algorithm} -- main algorithm set for this algorithm
  88. """
  89. current_algorithm = self
  90. parent_alrogithm = None
  91. # recursively find the main algorithm parent
  92. while current_algorithm._parent is not None:
  93. parent_alrogithm = current_algorithm._parent
  94. current_algorithm = current_algorithm._parent
  95. return parent_alrogithm
  96. def initRun(self):
  97. """
  98. Initialize the current solution and best solution using the `initialiser` function
  99. """
  100. self._currentSolution = self._initializer()
  101. # evaluate current solution
  102. self._currentSolution.evaluate(self._evaluator)
  103. # keep in memory best known solution (current solution)
  104. if self._bestSolution is None:
  105. self._bestSolution = self._currentSolution
  106. def increaseEvaluation(self):
  107. """
  108. Increase number of evaluation once a solution is evaluated for each dependant algorithm (parents hierarchy)
  109. """
  110. current_algorithm = self
  111. while current_algorithm is not None:
  112. current_algorithm._numberOfEvaluations += 1
  113. current_algorithm = current_algorithm._parent
  114. def getGlobalEvaluation(self):
  115. """Get the global number of evaluation (if inner algorithm)
  116. Returns:
  117. {int} -- current global number of evaluation
  118. """
  119. parent_algorithm = self.getParent()
  120. if parent_algorithm is not None:
  121. return parent_algorithm.getGlobalEvaluation()
  122. return self._numberOfEvaluations
  123. def getGlobalMaxEvaluation(self):
  124. """Get the global max number of evaluation (if inner algorithm)
  125. Returns:
  126. {int} -- current global max number of evaluation
  127. """
  128. parent_algorithm = self.getParent()
  129. if parent_algorithm is not None:
  130. return parent_algorithm.getGlobalMaxEvaluation()
  131. return self._maxEvaluations
  132. def stop(self):
  133. """
  134. Global stopping criteria (check for parents algorithm hierarchy too)
  135. """
  136. parent_algorithm = self.getParent()
  137. # based on global stopping creteria or on its own stopping critera
  138. if parent_algorithm is not None:
  139. return parent_algorithm._numberOfEvaluations >= parent_algorithm._maxEvaluations or self._numberOfEvaluations >= self._maxEvaluations
  140. return self._numberOfEvaluations >= self._maxEvaluations
  141. def evaluate(self, _solution):
  142. """
  143. Evaluate a solution using evaluator passed when intialize algorithm
  144. Args:
  145. solution: {Solution} -- solution to evaluate
  146. Returns:
  147. fitness score of solution which is not already evaluated or changed
  148. Note:
  149. if multi-objective problem this method can be updated using array of `evaluator`
  150. """
  151. return _solution.evaluate(self._evaluator)
  152. def update(self, _solution):
  153. """
  154. Apply update function to solution using specific `policy`
  155. Check if solution is valid after modification and returns it
  156. Args:
  157. solution: {Solution} -- solution to update using current policy
  158. Returns:
  159. {Solution} -- updated solution obtained by the selected operator
  160. """
  161. # two parameters are sent if specific crossover solution are wished
  162. sol = self._policy.apply(_solution)
  163. # compute fitness of new solution
  164. sol.evaluate(self._evaluator)
  165. if (sol.isValid(self._validator)):
  166. return sol
  167. else:
  168. logging.info("-- New solution is not valid %s" % sol)
  169. return _solution
  170. def isBetter(self, _solution):
  171. """
  172. Check if solution is better than best found
  173. Args:
  174. solution: {Solution} -- solution to compare with best one
  175. Returns:
  176. {bool} -- `True` if better
  177. """
  178. # depending of problem to solve (maximizing or minimizing)
  179. if self._maximise:
  180. if _solution.fitness() > self._bestSolution.fitness():
  181. return True
  182. else:
  183. if _solution.fitness() < self._bestSolution.fitness():
  184. return True
  185. # by default
  186. return False
  187. def run(self, _evaluations):
  188. """
  189. Run the specific algorithm following number of evaluations to find optima
  190. """
  191. # append number of max evaluation if multiple run called
  192. self._maxEvaluations += _evaluations
  193. # check if global evaluation is used or not
  194. if self.getParent() is not None and self.getGlobalEvaluation() != 0:
  195. # init number evaluations of inner algorithm depending of globalEvaluation
  196. # allows to restart from `checkpoint` last evaluation into inner algorithm
  197. rest = self.getGlobalEvaluation() % self._maxEvaluations
  198. self._numberOfEvaluations = rest
  199. else:
  200. self._numberOfEvaluations = 0
  201. logging.info("Run %s with %s evaluations" %
  202. (self.__str__(), _evaluations))
  203. def progress(self):
  204. """
  205. Log progress and apply callbacks if necessary
  206. """
  207. if len(self._callbacks) > 0:
  208. for callback in self._callbacks:
  209. callback.run()
  210. macop_progress(self.getGlobalEvaluation(),
  211. self.getGlobalMaxEvaluation())
  212. logging.info(f"-- {type(self).__name__} evaluation {self._numberOfEvaluations} of {self._maxEvaluations} ({((self._numberOfEvaluations / self._maxEvaluations) * 100):.2f}%) - BEST SCORE {self._bestSolution.fitness()}")
  213. def end(self):
  214. """Display end message into `run` method
  215. """
  216. print(macop_text(f'({type(self).__name__}) Found after {self._numberOfEvaluations} evaluations \n - {self._bestSolution}'))
  217. print(macop_line())
  218. def information(self):
  219. logging.info(f"-- Best {self._bestSolution} - SCORE {self._bestSolution.fitness()}")
  220. def __str__(self):
  221. return f"{type(self).__name__} using {type(self._bestSolution).__name__}"