"""Abstract Algorithm class used as basic algorithm implementation with some specific initialization
"""
# main imports
import logging
# Generic algorithm class
[docs]class Algorithm():
"""Algorithm class used as basic algorithm
Attributes:
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 optimization problem
currentSolution: {Solution} -- current solution managed for current evaluation
bestSolution: {Solution} -- best solution found so far during running algorithm
checkpoint: {Checkpoint} -- Checkpoint class implementation to keep track of algorithm and restart
parent: {Algorithm} -- parent algorithm reference in case of inner Algorithm instance (optional)
"""
def __init__(self,
_initalizer,
_evaluator,
_operators,
_policy,
_validator,
_maximise=True,
_parent=None):
self.initializer = _initalizer
self.evaluator = _evaluator
self.operators = _operators
self.policy = _policy
self.validator = _validator
self.checkpoint = None
self.bestSolution = None
# other parameters
self.parent = _parent # parent algorithm if it's sub algorithm
#self.maxEvaluations = 0 # by default
self.maximise = _maximise
self.initRun()
[docs] def addCheckpoint(self, _class, _every, _filepath):
"""Add checkpoint to algorithm specifying usefull parameters
Attributes:
_class: {class} -- Checkpoint class type
_every: {int} -- checkpoint frequency based on evaluations
_filepath: {str} -- file path where checkpoints will be saved
"""
self.checkpoint = _class(self, _every, _filepath)
[docs] def setCheckpoint(self, _checkpoint):
"""Set checkpoint instance directly
Attributes:
_checkpoint: {Checkpoint} -- checkpoint instance
"""
self.checkpoint = _checkpoint
[docs] def resume(self):
"""Resume algorithm using checkpoint instance
Raises:
ValueError: No checkpoint initialize (use `addCheckpoint` or `setCheckpoint` is you want to use this process)
"""
if self.checkpoint is None:
raise ValueError(
"Need to `addCheckpoint` or `setCheckpoint` is you want to use this process"
)
else:
print('Checkpoint loading is called')
self.checkpoint.load()
[docs] def initRun(self):
"""
Method which initialiazes or re-initializes the whole algorithm context: operators, current solution, best solution (by default current solution)
"""
# add track reference of algo into operator (keep an eye into best solution)
for operator in self.operators:
operator.setAlgo(self)
self.currentSolution = self.initializer()
# evaluate current solution
self.currentSolution.evaluate(self.evaluator)
# reinitialize policy
# if self.parent is not None:
# self.policy = globals()[type(self.policy).__name__]()
# keep in memory best known solution (current solution)
self.bestSolution = self.currentSolution
[docs] def increaseEvaluation(self):
"""
Increase number of evaluation once a solution is evaluated
"""
self.numberOfEvaluations += 1
if self.parent is not None:
self.parent.numberOfEvaluations += 1
[docs] def getGlobalEvaluation(self):
"""Get the global number of evaluation (if inner algorithm)
Returns:
{int} -- current global number of evaluation
"""
if self.parent is not None:
return self.parent.numberOfEvaluations
return self.numberOfEvaluations
[docs] def stop(self):
"""
Global stopping criteria (check for inner algorithm too)
"""
if self.parent is not None:
return self.parent.numberOfEvaluations >= self.parent.maxEvaluations or self.numberOfEvaluations >= self.maxEvaluations
return self.numberOfEvaluations >= self.maxEvaluations
[docs] def evaluate(self, solution):
"""
Returns:
fitness score of solution which is not already evaluated or changed
Note:
if multi-objective problem this method can be updated using array of `evaluator`
"""
return solution.evaluate(self.evaluator)
[docs] def update(self, solution):
"""
Apply update function to solution using specific `policy`
Check if solution is valid after modification and returns it
Returns:
{Solution} -- updated solution obtained by the selected operator
"""
# two parameters are sent if specific crossover solution are wished
sol = self.policy.apply(solution)
if (sol.isValid(self.validator)):
return sol
else:
logging.info("-- New solution is not valid %s" % sol)
return solution
[docs] def isBetter(self, solution):
"""
Check if solution is better than best found
Returns:
{bool} -- `True` if better
"""
# depending of problem to solve (maximizing or minimizing)
if self.maximise:
if self.evaluate(solution) > self.bestSolution.fitness():
return True
else:
if self.evaluate(solution) < self.bestSolution.fitness():
return True
# by default
return False
[docs] def run(self, _evaluations):
"""
Run the specific algorithm following number of evaluations to find optima
"""
self.maxEvaluations = _evaluations
self.initRun()
# check if global evaluation is used or not
if self.parent is not None and self.getGlobalEvaluation() != 0:
# init number evaluations of inner algorithm depending of globalEvaluation
# allows to restart from `checkpoint` last evaluation into inner algorithm
rest = self.getGlobalEvaluation() % self.maxEvaluations
self.numberOfEvaluations = rest
else:
self.numberOfEvaluations = 0
logging.info("Run %s with %s evaluations" %
(self.__str__(), _evaluations))
[docs] def progress(self):
"""
Log progress and apply checkpoint if necessary
"""
if self.checkpoint is not None:
self.checkpoint.run()
logging.info("-- %s evaluation %s of %s (%s%%) - BEST SCORE %s" %
(type(self).__name__, self.numberOfEvaluations,
self.maxEvaluations, "{0:.2f}".format(
(self.numberOfEvaluations) / self.maxEvaluations *
100.), self.bestSolution.fitness()))
def information(self):
logging.info("-- Best %s - SCORE %s" %
(self.bestSolution, self.bestSolution.fitness()))
def __str__(self):
return "%s using %s" % (type(self).__name__, type(
self.bestSolution).__name__)