123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412 |
- """Multi-objetive classes algorithm
- """
- import logging
- import math
- import numpy as np
- import sys
- from ..utils.color import macop_text, macop_line, macop_progress
- from .base import Algorithm
- from ..evaluators.multi import WeightedSum
- class MOEAD(Algorithm):
- """Multi-Ojective Evolutionary Algorithm with Scalar Decomposition
- Attributes:
- mu: {int} -- number of sub problems
- T: {[float]} -- number of neightbors for each sub problem
- nObjectives: {int} -- number of objectives (based of number evaluator)
- initalizer: {function} -- basic function strategy to initialize solution
- evaluator: {[function]} -- list of basic function in order to obtained fitness (multiple objectives)
- operators: {[Operator]} -- list of operator to use when launching algorithm
- policy: {Policy} -- Policy class implementation strategy to select operators
- validator: {function} -- basic function to check if solution is valid or not under some constraints
- maximise: {bool} -- specify kind of optimisation problem
- verbose: {bool} -- verbose or not information about the algorithm
- population: [{Solution}] -- population of solution, one for each sub problem
- pfPop: [{Solution}] -- pareto front population
- weights: [[{float}]] -- random weights used for custom mu sub problems
- callbacks: {[Callback]} -- list of Callback class implementation to do some instructions every number of evaluations and `load` when initializing algorithm
- """
- def __init__(self,
- mu,
- T,
- initalizer,
- evaluator,
- operators,
- policy,
- validator,
- maximise=True,
- parent=None,
- verbose=True):
-
- self._initializer = initalizer
- self._evaluator = evaluator
- self._operators = operators
- self._policy = policy
- self._validator = validator
- self._callbacks = []
-
- self._numberOfEvaluations = 0
- self._maxEvaluations = 0
- self._nObjectives = len(evaluator)
-
- self._parent = parent
-
- self._maximise = maximise
- self._verbose = verbose
-
- for operator in self._operators:
- operator.setAlgo(self)
-
- self._policy.setAlgo(self)
- if mu < T:
- raise ValueError('`mu` cannot be less than `T`')
- self._mu = mu
- self._T = T
-
- self.setNeighbors()
- weights = []
- if self._nObjectives == 2:
- for i in range(self._mu):
- angle = math.pi / 2 * i / (self._mu - 1)
- weights.append([math.cos(angle), math.sin(angle)])
- elif self._nObjectives >= 3:
-
- for i in range(self._mu):
- w_i = np.random.uniform(0, 1, self._nObjectives)
- weights.append(w_i / sum(w_i))
- else:
- raise ValueError('Unvalid number of objectives')
- self._weights = weights
- self._subProblems = []
- for i in range(self._mu):
-
- sub_evaluator = WeightedSum(data={'evaluators': evaluator, 'weights': weights[i]})
-
-
- subProblem = MOSubProblem(i, weights[i],
- initalizer, sub_evaluator,
- operators.copy(), policy, validator,
- maximise, self)
- self._subProblems.append(subProblem)
- self._population = [None for n in range(self._mu)]
- self._pfPop = []
-
- if self._maximise:
- self._refPoint = [0 for _ in range(self._nObjectives)]
- else:
- self._refPoint = [
- sys.float_info.max for _ in range(self._nObjectives)
- ]
- def initRun(self):
- """
- Method which initialiazes or re-initializes the whole algorithm context specifically for MOEAD
- """
-
- pass
- def run(self, evaluations):
- """
- Run the local search algorithm
- Args:
- evaluations: {int} -- number of Local search evaluations
-
- Returns:
- {Solution} -- best solution found
- """
-
- super().run(evaluations)
-
- self.resume()
-
- for i in range(self._mu):
- if self._subProblems[i]._bestSolution is None:
- self._subProblems[i].run(1)
- self._population[i] = self._subProblems[i]._bestSolution
-
- if len(self._pfPop) == 0:
- for i in range(self._mu):
- self._pfPop.append(self._subProblems[i]._bestSolution)
-
- while not self.stop():
- for i in range(self._mu):
-
- self._subProblems[i].run(1)
- spBestSolution = self._subProblems[i]._bestSolution
- self.updateRefPoint(spBestSolution)
-
- improvment = False
- for j in self._neighbors[i]:
- if spBestSolution.fitness() > self._subProblems[j]._bestSolution.fitness():
-
- newSolution = spBestSolution.clone()
-
- self._subProblems[j].evaluate(newSolution)
- self._subProblems[j]._bestSolution = newSolution
-
- self._population[j] = newSolution
- improvment = True
-
- if improvment:
- self._pfPop.append(spBestSolution)
-
- self._pfPop = self.paretoFront(self._pfPop)
-
- self.progress()
-
- if self.stop():
- break
- logging.info(f"End of {type(self).__name__}, best solution found {self._population}")
- self.end()
- return self._pfPop
- def progress(self):
- """
- Log progress and apply callbacks if necessary
- """
- if len(self._callbacks) > 0:
- for callback in self._callbacks:
- callback.run()
- macop_progress(self.getGlobalEvaluation(), self.getGlobalMaxEvaluation())
- logging.info(f"-- {type(self).__name__} evaluation {self._numberOfEvaluations} of {self._maxEvaluations} ({((self._numberOfEvaluations) / self._maxEvaluations * 100.):.2f}%)")
- def setNeighbors(self):
- dmin = dmax = 0
- if self._T % 2 == 1:
- dmin = -int(self._T / 2)
- dmax = int(self._T / 2) + 1
- else:
- dmin = -int(self._T / 2) + 1
- dmax = +self._T / 2
-
- self._neighbors = [[] for n in range(self._mu)]
- for direction in range(0, -dmin):
- for i in range(self._T):
- self._neighbors[direction].append(i)
- for direction in range(-dmin, self._mu - dmax):
- for i in range(direction + dmin, direction + dmax):
- self._neighbors[direction].append(i)
- for direction in range(self._mu - dmax, self._mu):
- for i in range(self._mu - self._T, self._mu):
- self._neighbors[direction].append(i)
- def updateRefPoint(self, solution):
- if self._maximise:
- for i in range(len(self._evaluator)):
- if solution._scores[i] > self._refPoint[i]:
- self._refPoint[i] = solution._scores[i]
- else:
- for i in range(len(self._evaluator)):
- if solution.scores[i] < self._refPoint[i]:
- self._refPoint[i] = solution._scores[i]
- def paretoFront(self, population):
- paFront = []
- indexes = []
- nObjectives = len(self._evaluator)
- nSolutions = len(population)
-
- for i in range(nSolutions):
- for j in range(nSolutions):
- if j in indexes:
- continue
-
- if all([
- population[i]._data[k] == population[j]._data[k]
- for k in range(len(population[i]._data))
- ]):
- continue
- nDominated = 0
-
- for k in range(len(self._evaluator)):
- if self._maximise:
- if population[i]._scores[k] < population[j]._scores[k]:
- nDominated += 1
- else:
- if population[i]._scores[k] > population[j]._scores[k]:
- nDominated += 1
- if nDominated == nObjectives:
- indexes.append(i)
- break
-
- for i in range(nSolutions):
- if i not in indexes:
- paFront.append(population[i])
- return paFront
- def end(self):
- """Display end message into `run` method
- """
- print(macop_text(f'({type(self).__name__}) Found after {self._numberOfEvaluations} evaluations'))
- for i, solution in enumerate(self._pfPop):
- print(f' - [{i}] {solution._scores} : {solution}')
- print(macop_line())
- def information(self):
- logging.info("-- Pareto front :")
- for i, solution in enumerate(self._pfPop):
- logging.info(f"-- {i}] SCORE {solution._scores} - {solution}")
- def __str__(self):
- return f"{type(self).__name__} using {type(self._population).__name__}"
- class MOSubProblem(Algorithm):
- """Specific MO sub problem used into MOEAD
- Attributes:
- index: {int} -- sub problem index
- weights: {[float]} -- sub problems objectives weights
- initalizer: {function} -- basic function strategy to initialize solution
- evaluator: {function} -- basic function in order to obtained fitness (mono or multiple objectives)
- operators: {[Operator]} -- list of operator to use when launching algorithm
- policy: {Policy} -- Policy class implementation strategy to select operators
- validator: {function} -- basic function to check if solution is valid or not under some constraints
- maximise: {bool} -- specify kind of optimisation problem
- verbose: {bool} -- verbose or not information about the algorithm
- currentSolution: {Solution} -- current solution managed for current evaluation
- bestSolution: {Solution} -- best solution found so far during running algorithm
- callbacks: {[Callback]} -- list of Callback class implementation to do some instructions every number of evaluations and `load` when initializing algorithm
- """
- def __init__(self,
- index,
- weights,
- initalizer,
- evaluator,
- operators,
- policy,
- validator,
- maximise=True,
- parent=None,
- verbose=True):
- super().__init__(initalizer, evaluator, operators, policy,
- validator, maximise, parent)
- self._index = index
- self._weights = weights
- self._verbose = verbose
- def run(self, evaluations):
- """
- Run the local search algorithm
- Args:
- evaluations: {int} -- number of evaluations
-
- Returns:
- {Solution} -- best solution found
- """
-
- super().run(evaluations)
-
- if self._bestSolution is None:
- self.initRun()
-
- for op in self._operators:
- op.setAlgo(self)
- for _ in range(evaluations):
-
- self._policy.setAlgo(self)
-
- newSolution = self.update(self._bestSolution)
-
- if self.isBetter(newSolution):
- self._bestSolution = newSolution
-
- self.increaseEvaluation()
- self.progress()
-
- if self.stop():
- break
- logging.info(f"---- Current {newSolution} - SCORE {newSolution.fitness()}")
- logging.info(f"End of {type(self).__name__}, best solution found {self._bestSolution}")
- return self._bestSolution
|