-
Notifications
You must be signed in to change notification settings - Fork 7
/
Application.py
256 lines (195 loc) · 10.4 KB
/
Application.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
# -*- coding: utf-8 -*-
# MLC (Machine Learning Control): A genetic algorithm library to solve chaotic problems
# Copyright (C) 2015-2017, Thomas Duriez ([email protected])
# Copyright (C) 2015, Adrian Durán ([email protected])
# Copyright (C) 2015-2017, Ezequiel Torres Feyuk ([email protected])
# Copyright (C) 2016-2017, Marco Germano Zbrun ([email protected])
# Copyright (C) 2016-2017, Raúl Lopez Skuba ([email protected])
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>
import numpy as np
import MLC.Log.log as lg
from MLC.Common.PreevaluationManager import PreevaluationManager
from MLC.Common.Operations import Operations
from MLC.db.mlc_repository import MLCRepository
from MLC.Log.log import set_logger
from MLC.mlc_parameters.mlc_parameters import Config
from MLC.Population.Creation.CreationFactory import CreationFactory
from MLC.Population.Evaluation.EvaluatorFactory import EvaluatorFactory
from MLC.Simulation import Simulation
class MLC_CALLBACKS:
ON_START = 0
ON_EVALUATE = 1
ON_NEW_GENERATION = 2
ON_FINISH = 3
class Application(object):
def __init__(self, simulation, callbacks={}, gen_creator=None):
self._config = Config.get_instance()
# Reload the Operations supported
Operations.get_instance(reload_operations=True)
self._simulation = simulation
self._mlc_repository = MLCRepository.get_instance()
# Set logger mode of the App
set_logger(self._config.get('LOGGING', 'logmode'))
self._simulation = simulation
self._project_validations()
# callbacks configuration
self.__callbacks_manager = MLCCallbacksManager()
# bad values and duplicates
self.__badvalues_elim = self._config.get('EVALUATOR', 'badvalues_elim')
# Gen creator
if gen_creator is None:
gen_method = self._config.get('GP', 'generation_method')
self._gen_creator = CreationFactory.make(gen_method)
else:
self._gen_creator = gen_creator
# Gen evaluator
ev_method = self._config.get('EVALUATOR', 'evaluation_method')
self._evaluator = EvaluatorFactory.make(ev_method, self.__callbacks_manager)
self._look_for_duplicates = self._config.getboolean('OPTIMIZATION', 'lookforduplicates')
# callbacks for the MLC application
if MLC_CALLBACKS.ON_START in callbacks:
self.__callbacks_manager.subscribe(MLC_CALLBACKS.ON_START,
callbacks[MLC_CALLBACKS.ON_START])
if MLC_CALLBACKS.ON_EVALUATE in callbacks:
self.__callbacks_manager.subscribe(MLC_CALLBACKS.ON_EVALUATE,
callbacks[MLC_CALLBACKS.ON_EVALUATE])
if MLC_CALLBACKS.ON_NEW_GENERATION in callbacks:
self.__callbacks_manager.subscribe(MLC_CALLBACKS.ON_NEW_GENERATION,
callbacks[MLC_CALLBACKS.ON_NEW_GENERATION])
if MLC_CALLBACKS.ON_FINISH in callbacks:
self.__callbacks_manager.subscribe(MLC_CALLBACKS.ON_FINISH,
callbacks[MLC_CALLBACKS.ON_FINISH])
# add callback to show best individual
self.__callbacks_manager.subscribe(MLC_CALLBACKS.ON_NEW_GENERATION, self.show_best)
self.__display_best = True
def _set_numpy_parameters(self):
# Set printable resolution (don't alter numpy interval resolution)
np.set_printoptions(precision=9)
# Show full arrays, no matter what size do they have
np.set_printoptions(threshold=np.inf)
# Don't show scientific notation
np.set_printoptions(suppress=True)
def go(self, to_generation, from_generation=None, display_best=False):
"""
Start MLC2 problem solving (MLC2 Toolbox)
:param to_generation: creates (if necessary) the population, evaluate
and evolve it until to_generation evaluated generations are
obtained.
:param from_generation: first generation must be evolved from
from_generation, takes last generation as default value.
:param display_best: displays the best individual if implemented
in the evaluation function at the end of each generation evaluation.
:return:
"""
self._set_numpy_parameters()
self.__display_best = display_best
if from_generation is None:
from_generation = self._mlc_repository.count_population()
lg.logger_.info("Running MLC from generation %s to %s" % (from_generation, to_generation))
if from_generation < self._mlc_repository.count_population():
lg.logger_.info("Generations %s to %s discarded" % (from_generation + 1, self._mlc_repository.count_population()))
self._mlc_repository.remove_population_from(from_generation + 1)
# emit app start event
self.__callbacks_manager.on_event(MLC_CALLBACKS.ON_START)
# First generation must be generated from scratch
if self._mlc_repository.count_population() == 0:
lg.logger_.info("Creating and filling first generation")
last_population = Simulation.create_empty_population_for(1)
last_population.fill(self._gen_creator)
self.evaluate_population(last_population, 1)
self._mlc_repository.add_population(last_population)
# emit new generation event
self.__callbacks_manager.on_event(MLC_CALLBACKS.ON_NEW_GENERATION, 1)
lg.logger_.info("Population created. Number: %s - Size: %s" % (1, last_population.get_size()))
while self._mlc_repository.count_population() < to_generation:
last_generation = self._mlc_repository.count_population()
last_population = self._mlc_repository.get_population(last_generation)
# obtain the next generation by evolving the lastone
lg.logger_.info("Evolving to Population %s using population %s" % (last_generation + 1, last_generation))
next_population = Simulation.create_empty_population_for(last_generation + 1)
next_population = last_population.evolve(next_population)
# continue with evolve if there are duplicated individuals
if self._look_for_duplicates:
while next_population.remove_duplicates() > 0:
next_population = last_population.evolve(next_population)
# evaluate population
self.evaluate_population(next_population, last_generation)
lg.logger_.info("Population created. Number: %s - Size: %s" % (last_generation + 1, next_population.get_size()))
self._mlc_repository.add_population(next_population)
# emit new generation event
self.__callbacks_manager.on_event(MLC_CALLBACKS.ON_NEW_GENERATION, last_generation + 1)
lg.logger_.info("MLC Simulation Finished")
# emit app finish event
self.__callbacks_manager.on_event(MLC_CALLBACKS.ON_FINISH)
def get_simulation(self):
return self._simulation
def evaluate_population(self, population, generation_number):
"""
Evolves the population. (MLC2 Toolbox)
OBJ.EVALUATE_POPULATION launches the evaluation method,
and updates the MLC2 object.
The evaluation algorithm is implemented in the MLCpop class.
"""
# First evaluation
population.evaluate(self._evaluator)
# Remove bad individuals
if self._duplicates_must_be_removed(generation_number):
while population.remove_bad_individuals():
# There are bad individuals, recreate the population
population.fill(self._gen_creator)
population.evaluate(self._evaluator)
population.sort()
# Enforce reevaluation
if self._config.getboolean('EVALUATOR', 'ev_again_best'):
ev_again_times = self._config.getint('EVALUATOR', 'ev_again_times')
for i in range(1, ev_again_times):
ev_again_nb = self._config.getint('EVALUATOR', 'ev_again_nb')
population.evaluate(self._evaluator)
population.sort()
def _duplicates_must_be_removed(self, generation_number):
if self.__badvalues_elim == "all":
return True
elif self.__badvalues_elim == "first":
return (generation_number == 1)
return False
def show_best(self, generation_number):
if self.__display_best:
show_all = self._config.getboolean('BEHAVIOUR', 'showeveryitbest')
stop_on_graph = self._config.getboolean('BEHAVIOUR', 'stopongraph')
population = self._mlc_repository.get_population(generation_number)
best_index, best_indiv, cost = population.get_best_individual()
EvaluatorFactory.get_callback().show_best(index=best_index,
indiv=best_indiv,
cost=cost,
generation=generation_number,
block=stop_on_graph)
def _project_validations(self):
# Check that the evaluation and preevaluation modules can be loaded
EvaluatorFactory.get_callback()
PreevaluationManager.get_callback()
class MLCCallbacksManager:
def __init__(self):
self.__callbacks = {}
def subscribe(self, event_type, callback):
if event_type not in self.__callbacks:
self.__callbacks[event_type] = []
if not isinstance(callback, list):
callback = [callback]
self.__callbacks[event_type].extend(callback)
def on_event(self, event_type, *args, **kwargs):
if event_type not in self.__callbacks:
return
for callback in self.__callbacks[event_type]:
callback(*args, **kwargs)