source: framspy/evolalg/base/experiment_niching_abc.py @ 1190

Last change on this file since 1190 was 1190, checked in by Maciej Komosinski, 16 months ago

Added the "evolalg" module for evolutionary optimization

File size: 8.9 KB
Line 
1import time
2from abc import ABC, abstractmethod
3from tkinter import W
4
5import numpy as np
6from deap import base, tools
7from deap.tools.emo import assignCrowdingDist
8
9from ..constants import BAD_FITNESS
10from ..structures.individual import Individual
11from .experiment_abc import ExperimentABC
12from .remove_diagonal import remove_diagonal
13
14
15class DeapFitness(base.Fitness):
16    weights = (1, 1)
17
18    def __init__(self, *args, **kwargs):
19        super(DeapFitness, self).__init__(*args, **kwargs)
20
21
22class ExperimentNiching(ExperimentABC, ABC):
23    fit: str = "niching"
24    normalize: str = "None"
25    archive_size: int = None
26
27    def __init__(self, fit, normalize, popsize, hof_size, save_only_best=True, knn_niching=5, knn_nslc=10, archive_size=0) -> None:
28        ExperimentABC.__init__(self,popsize=popsize, hof_size=hof_size, save_only_best=save_only_best)
29        self.fit = fit
30        self.knn_niching = knn_niching
31        self.knn_nslc = knn_nslc
32        self.normalize = normalize
33        self.archive_size=archive_size
34        if popsize < self.knn_niching:
35            self.knn_niching = popsize - 2
36        if popsize < self.knn_nslc:
37            self.knn_nslc = popsize - 2
38
39    def transform_indexes(self, i, index_array):
40        return [x+1 if x >= i else x for x in index_array]
41
42    def normalize_dissim(self, dissim_matrix):
43        dissim_matrix = remove_diagonal(np.array(dissim_matrix))
44        if self.normalize == "none":
45            return dissim_matrix
46        elif self.normalize == "max":
47            divide_by = np.max(dissim_matrix)
48        elif self.normalize == "sum":
49            divide_by = np.sum(dissim_matrix)
50        else:
51            raise Exception(f"Wrong normalization method,", self.normalize)
52        if divide_by != 0:
53            return dissim_matrix/divide_by
54        else:
55            return dissim_matrix
56
57    def do_niching(self, population_structures):
58        population_archive = population_structures.population + population_structures.archive
59        dissim_matrix = self.dissimilarity(population_archive)
60        if "knn" not in self.fit:
61            dissim_list = np.mean(self.normalize_dissim(dissim_matrix), axis=1)
62        else:
63            dissim_list = np.mean(np.partition(
64                self.normalize_dissim(dissim_matrix), self.knn_niching)[:, :self.knn_niching], axis=1)
65
66        if "niching" in self.fit:
67            for i, d in zip(population_archive, dissim_list):
68                i.fitness = i.rawfitness * d
69        elif "novelty" in self.fit:
70            for i, d in zip(population_archive, dissim_list):
71                i.fitness = d
72        else:
73            raise Exception("Wrong fit type: ", self.fit,
74                            f" chose correct one or implement new behaviour")
75        population_structures.update_archive(dissim_matrix, population_archive)
76
77    def do_nsga2_dissim(self, population):
78        dissim_matrix = self.dissimilarity(population)
79        dissim_list = np.mean(self.normalize_dissim(dissim_matrix), axis=1)
80        for i, d in zip(population, dissim_list):
81            i.fitness = DeapFitness(tuple((d, i.rawfitness)))
82
83    def do_nslc_dissim(self, population):
84        dissim_matrix = self.dissimilarity(population)
85        normalized_matrix = self.normalize_dissim(dissim_matrix)
86        for i in range(len(normalized_matrix)):
87            temp_dissim = normalized_matrix[i]
88            index_array = np.argpartition(temp_dissim, kth=self.knn_nslc, axis=-1)[:self.knn_nslc]
89            dissim_value = np.mean(np.take_along_axis(
90                temp_dissim, index_array, axis=-1))
91            temp_fitness = population[i].rawfitness
92            population_of_most_different = list(
93                map(population.__getitem__, self.transform_indexes(i, index_array)))
94            temp_ind_fit = sum(
95                [1 for ind in population_of_most_different if ind.rawfitness < temp_fitness])
96            population[i].fitness = DeapFitness(
97                tuple((dissim_value, temp_ind_fit)))
98
99    def make_new_population_nsga2(self, population, prob_mut, prob_xov):
100        N = len(population)
101        expected_mut = int(N * prob_mut)
102        expected_xov = int(N * prob_xov)
103        assignCrowdingDist(population)
104        offspring = tools.selTournamentDCD(population, N)
105
106        def addGenotypeIfValid(ind_list, genotype):
107            new_individual = Individual()
108            new_individual.set_and_evaluate(genotype, self.evaluate)
109            if new_individual.fitness is not BAD_FITNESS:  # this is how we defined BAD_FITNESS in frams_evaluate()
110                ind_list.append(new_individual)
111
112        counter = 0
113
114        def get_indyvidual(pop, c):
115            if c < len(pop):
116                ind = pop[c]
117                c += 1
118                return ind, c
119            else:
120                c = 0
121                ind = pop[c]
122                c += 1
123                return ind, c
124
125        newpop = []
126        while len(newpop) < expected_mut:
127            ind, counter = get_indyvidual(offspring, counter)
128            addGenotypeIfValid(newpop, self.mutate(ind.genotype))
129
130        # adding valid crossovers of selected individuals...
131        while len(newpop) < expected_mut + expected_xov:
132            ind1, counter = get_indyvidual(offspring, counter)
133            ind2, counter = get_indyvidual(offspring, counter)
134            addGenotypeIfValid(newpop, self.cross_over(
135                ind1.genotype, ind2.genotype))
136
137        # select clones to fill up the new population until we reach the same size as the input population
138        while len(newpop) < len(population):
139            ind, counter = get_indyvidual(offspring, counter)
140            newpop.append(Individual().copyFrom(ind))
141
142        pop_offspring = population+newpop
143        print(len(pop_offspring))
144        if self.fit == "nslc":
145            self.do_nslc_dissim(pop_offspring)
146        elif self.fit == "nsga2":
147            self.do_nsga2_dissim(pop_offspring)
148        out_pop = tools.selNSGA2(pop_offspring, len(population))
149        return out_pop
150
151    def evolve(self, hof_savefile, generations, initialgenotype, pmut, pxov, tournament_size):
152        file_name = self.get_state_filename(hof_savefile)
153        state = self.load_state(file_name)
154        if state is not None:  # loaded state from file
155            # saved generation has been completed, start with the next one
156            self.current_generation += 1
157            print("...Resuming from saved state: population size = %d, hof size = %d, stats size = %d, archive size = %d, generation = %d/%d" % (len(self.population_structures.population), len(self.hof),
158                                                                                                                                                 len(self.stats),  (len(self.population_structures.archive)), self.current_generation, generations))  # self.current_generation (and g) are 0-based, parsed_args.generations is 1-based
159        else:
160            self.initialize_evolution(self.genformat, initialgenotype)
161
162        time0 = time.process_time()
163        for g in range(self.current_generation, generations):
164            if self.fit != "raw" and self.fit != "nsga2" and self.fit != "nslc":
165                self.do_niching(self.population_structures)
166
167            if type(self.population_structures.population[0].fitness) == DeapFitness:
168                self.population_structures.population = self.make_new_population_nsga2(
169                    self.population_structures.population, pmut, pxov)
170            else:
171                self.population_structures.population = self.make_new_population(
172                    self.population_structures.population, pmut, pxov, tournament_size)
173
174            self.update_stats(g, self.population_structures.population)
175
176            if hof_savefile is not None:
177                self.current_generation = g
178                self.time_elapsed += time.process_time() - time0
179                self.save_state(file_name)
180        if hof_savefile is not None:
181            self.save_genotypes(hof_savefile)
182        return self.population_structures.population, self.stats
183
184    @staticmethod
185    def get_args_for_parser():
186        parser = ExperimentABC.get_args_for_parser()
187        parser.add_argument("-dissim",type= int, default= "frams",
188                   help="Dissimilarity measure type. Availible -2:emd, -1:lev, 1:frams1 (default}, 2:frams2")
189        parser.add_argument("-fit",type= str, default= "raw",
190                        help="Fitness type, availible  types: niching, novelty, nsga2, nslc and raw (default}")
191        parser.add_argument("-archive",type= int, default= 50,
192                            help="Maximum archive size")
193        parser.add_argument("-normalize",type= str, default= "max",
194                            help="What normalization use for dissimilarity matrix, max (default}, sum and none")
195        parser.add_argument("-knn",type= int, default= 0,
196                        help="Nearest neighbours parameter for local novelty/niching, if knn==0 global is performed.Default:0")
197        return parser
198       
199    @abstractmethod
200    def dissimilarity(self, population: list):
201        pass
Note: See TracBrowser for help on using the repository browser.