123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149 |
- # main imports
- import logging
- # Generic algorithm class
- class Algorithm():
- def __init__(self, _initalizer, _evaluator, _operators, _policy, _validator, _maximise=True, _parent=None):
- """
- Initialize all usefull parameters for problem to solve
- """
- self.initializer = _initalizer
- self.evaluator = _evaluator
- self.operators = _operators
- self.validator = _validator
- self.policy = _policy
- self.checkpoint = None
- # other parameters
- self.parent = _parent # parent algorithm if it's sub algorithm
- self.maxEvalutations = 0 # by default
- self.maximise = _maximise
- self.initRun()
- def addCheckpoint(self, _class, _every, _filepath):
- self.checkpoint = _class(self, _every, _filepath)
-
- def setCheckpoint(self, _checkpoint):
- self.checkpoint = _checkpoint
- def resume(self):
- 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()
- def initRun(self):
- """
- Reinit the whole variables
- """
- self.currentSolution = self.initializer()
-
- # evaluate current solution
- self.currentSolution.evaluate(self.evaluator)
- # keep in memory best known solution (current solution)
- self.bestSolution = self.currentSolution
- self.numberOfEvaluations = 0
- def increaseEvaluation(self):
- self.numberOfEvaluations += 1
- if self.parent is not None:
- self.parent.numberOfEvaluations += 1
-
- def getGlobalEvaluation(self):
- if self.parent is not None:
- return self.parent.numberOfEvaluations
- return self.numberOfEvaluations
- 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)
- def update(self, solution, secondSolution=None):
- """
- Apply update function to solution using specific `policy`
- Check if solution is valid after modification and returns it
- Returns:
- updated solution
- """
- # two parameters are sent if specific crossover solution are wished
- sol = self.policy.apply(solution, secondSolution)
- if(sol.isValid(self.validator)):
- return sol
- else:
- logging.info("-- New solution is not valid %s" % sol)
- return solution
- def isBetter(self, solution):
- """
- Check if solution is better than best found
- Returns:
- `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
- def run(self, _evaluations):
- """
- Run the specific algorithm following number of evaluations to find optima
- """
- self.maxEvalutations = _evaluations
- self.initRun()
- logging.info("Run %s with %s evaluations" % (self.__str__(), _evaluations))
- def progress(self):
- 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.maxEvalutations, "{0:.2f}".format((self.numberOfEvaluations) / self.maxEvalutations * 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__)
|