Thesis-NoiseDectection-rfe-attributes/optimization/algorithms/Algorithm.py

# 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.maxEvaluations = 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
        

    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 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


    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.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))


    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.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__)