Last active
December 17, 2015 00:59
-
-
Save cmd-ntrf/5525264 to your computer and use it in GitHub Desktop.
DEAP - Proposal for a replacement to the Statistics and EvolutionLogger objects. The Database object combine both object functionalities. It requires numpy for statistics computation.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
try: | |
import numpy | |
except ImportError: | |
import platform | |
if platform.python_implementation() == "PyPy": | |
import numpypy as numpy | |
else: | |
raise ImportError("DEAP requires Numpy.") | |
from collections import OrderedDict | |
import pprint | |
import curses | |
def identity(obj): | |
return obj | |
class Interface(object): | |
def __init__(self, database): | |
self.db = database | |
self.screen = curses.initscr() | |
self.screen.scrollok(True) | |
self.screen.clear() | |
def refresh(self): | |
self.screen.addstr(0, 0, str(self.db)) | |
self.screen.refresh() | |
def wait(self): | |
self.screen.getkey() | |
curses.endwin() | |
class Database(list): | |
def __init__(self, key=identity): | |
self.key = key | |
self.functions = {} | |
self.axis = 0 | |
self.header = None | |
self.buffindex = 0 | |
def __getstate__(self): | |
state = {} | |
state['axis'] = self.axis | |
state['functions'] = self.functions | |
state['header'] = self.header | |
# Most key cannot be pickled in Python 2 | |
state['key'] = None | |
return state | |
def __setstate__(self, state): | |
self.__init__(state['key']) | |
self.axis = state['axis'] | |
self.functions = state['functions'] | |
self.header = state['header'] | |
def register(self, name, function): | |
self.functions[name] = function | |
def select(self, *names): | |
if len(names) == 1: | |
return numpy.vstack((entry[names[0]] for entry in self)) | |
result = {} | |
for name in names: | |
result[name] = numpy.vstack((entry[name] for entry in self)) | |
return result | |
def append(self, data=[], **kargs): | |
if not self.key: | |
raise AttributeError('It is required to set a key after unpickling a %s object.' % self.__class__.__name__) | |
if not self.header: | |
self.header = kargs.keys() + self.functions.keys() | |
values = numpy.array([self.key(elem) for elem in data]) | |
entry = OrderedDict.fromkeys(self.header, "") | |
for key, func in self.functions.iteritems(): | |
entry[key] = func(values, self.axis) | |
entry.update(kargs) | |
list.append(self, entry) | |
@property | |
def stream(self): | |
"""Retrieve the formated unstreamed entries of the database including | |
the headers.""" | |
startindex, self.buffindex = self.buffindex, len(self) | |
return self.__str__(startindex) | |
def __str__(self, startindex=0): | |
columns_len = map(len, self.header) | |
str_matrix = [map(str, (line.get(name, "") for name in self.header)) for line in self[startindex:]] | |
for line in str_matrix: | |
for i, column in enumerate(line): | |
columns_len[i] = max(columns_len[i], len(column)) | |
template = "\t".join(("{:<%i}" % i for i in columns_len)) | |
text = [] | |
if startindex == 0: | |
text.append(template.format(*self.header)) | |
for line in str_matrix: | |
text.append(template.format(*line)) | |
return "\n".join(text) | |
if __name__ == "__main__": | |
db = Database() | |
db.register("mean", numpy.mean) | |
db.register("max", numpy.max) | |
db.header = ['gen', 'deme', 'mean', 'max'] | |
print db | |
import time | |
interface = Interface(db) | |
db.append([[0, 1], [2, 3]], gen=0, deme=0) | |
interface.refresh() | |
time.sleep(1) | |
db.append([[0, 1], [4, 5]], gen=0, deme=1) | |
interface.refresh() | |
time.sleep(1) | |
db.append([[0, 1], [2, 3]], gen=1, deme=0) | |
interface.refresh() | |
time.sleep(1) | |
db.append([[0, 1], [6, 7]], gen=1, deme=1) | |
interface.refresh() | |
time.sleep(1) | |
interface.wait() | |
print db.select("mean", "deme") |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import array | |
import random | |
import numpy | |
from itertools import chain | |
from datalogger import Database, Interface | |
from deap import algorithms | |
from deap import base | |
from deap import creator | |
from deap import tools | |
creator.create("FitnessMax", base.Fitness, weights=(1.0, 1.0)) | |
creator.create("Individual", array.array, typecode='b', fitness=creator.FitnessMax) | |
toolbox = base.Toolbox() | |
# Attribute generator | |
toolbox.register("attr_bool", random.randint, 0, 1) | |
# Structure initializers | |
toolbox.register("individual", tools.initRepeat, creator.Individual, | |
toolbox.attr_bool, 100) | |
toolbox.register("population", tools.initRepeat, list, toolbox.individual) | |
def evalOneMax(individual): | |
return sum(individual), | |
toolbox.register("evaluate", evalOneMax) | |
toolbox.register("mate", tools.cxTwoPoints) | |
toolbox.register("mutate", tools.mutFlipBit, indpb=0.05) | |
toolbox.register("select", tools.selTournament, tournsize=3) | |
toolbox.register("migrate", tools.migRing, k=5, selection=tools.selBest, | |
replacement=random.sample) | |
def main(): | |
random.seed(64) | |
NBR_DEMES = 3 | |
MU = 300 | |
NGEN = 50 | |
CXPB = 0.5 | |
MUTPB = 0.2 | |
MIG_RATE = 5 | |
demes = [toolbox.population(n=MU) for _ in range(NBR_DEMES)] | |
hof = tools.HallOfFame(1) | |
db = Database(lambda ind: ind.fitness.values) | |
db.register("avg", numpy.mean) | |
db.register("std", numpy.std) | |
db.register("min", numpy.min) | |
db.register("max", numpy.max) | |
db.header = ['gen', 'deme', 'evals', 'avg', 'max', 'min', 'std'] | |
interface = Interface(db) | |
for idx, deme in enumerate(demes): | |
for ind in deme: | |
ind.fitness.values = toolbox.evaluate(ind) | |
hof.update(deme) | |
for i, deme in enumerate(demes): | |
db.append(deme, gen=0, deme=i, evals=len(deme)) | |
interface.refresh() | |
gen = 1 | |
while gen <= NGEN: | |
for idx, deme in enumerate(demes): | |
deme[:] = toolbox.select(deme, len(deme)) | |
deme[:] = algorithms.varAnd(deme, toolbox, cxpb=CXPB, mutpb=MUTPB) | |
invalid_ind = [ind for ind in deme if not ind.fitness.valid] | |
for ind in invalid_ind: | |
ind.fitness.values = toolbox.evaluate(ind) | |
db.append(deme, gen=gen, deme=idx, evals=len(invalid_ind)) | |
hof.update(deme) | |
db.append(chain(*demes), gen=gen) | |
interface.refresh() | |
if gen % MIG_RATE == 0: | |
toolbox.migrate(demes) | |
gen += 1 | |
interface.wait() | |
return demes, db, hof | |
if __name__ == "__main__": | |
main() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import array | |
import random | |
import numpy | |
from itertools import chain | |
from datalogger import Database, Interface | |
from deap import algorithms | |
from deap import base | |
from deap import creator | |
from deap import tools | |
creator.create("FitnessMax", base.Fitness, weights=(1.0,)) | |
creator.create("Individual", array.array, typecode='b', fitness=creator.FitnessMax) | |
toolbox = base.Toolbox() | |
# Attribute generator | |
toolbox.register("attr_bool", random.randint, 0, 1) | |
# Structure initializers | |
toolbox.register("individual", tools.initRepeat, creator.Individual, | |
toolbox.attr_bool, 100) | |
toolbox.register("population", tools.initRepeat, list, toolbox.individual) | |
def evalOneMax(individual): | |
return sum(individual), | |
toolbox.register("evaluate", evalOneMax) | |
toolbox.register("mate", tools.cxTwoPoints) | |
toolbox.register("mutate", tools.mutFlipBit, indpb=0.05) | |
toolbox.register("select", tools.selTournament, tournsize=3) | |
toolbox.register("migrate", tools.migRing, k=5, selection=tools.selBest, | |
replacement=random.sample) | |
def main(): | |
random.seed(64) | |
NBR_DEMES = 3 | |
MU = 300 | |
NGEN = 50 | |
CXPB = 0.5 | |
MUTPB = 0.2 | |
MIG_RATE = 5 | |
demes = [toolbox.population(n=MU) for _ in range(NBR_DEMES)] | |
hof = tools.HallOfFame(1) | |
db = Database(lambda ind: ind.fitness.values) | |
db.register("avg", numpy.mean) | |
db.register("std", numpy.std) | |
db.register("min", numpy.min) | |
db.register("max", numpy.max) | |
db.header = ['gen', 'deme', 'evals', 'avg', 'max', 'min'] | |
for idx, deme in enumerate(demes): | |
for ind in deme: | |
ind.fitness.values = toolbox.evaluate(ind) | |
hof.update(deme) | |
for i, deme in enumerate(demes): | |
db.append(deme, gen=0, deme=i, evals=len(deme)) | |
print db.stream | |
gen = 1 | |
while gen <= NGEN: | |
for idx, deme in enumerate(demes): | |
deme[:] = toolbox.select(deme, len(deme)) | |
deme[:] = algorithms.varAnd(deme, toolbox, cxpb=CXPB, mutpb=MUTPB) | |
invalid_ind = [ind for ind in deme if not ind.fitness.valid] | |
for ind in invalid_ind: | |
ind.fitness.values = toolbox.evaluate(ind) | |
db.append(deme, gen=gen, deme=idx, evals=len(invalid_ind)) | |
print db.stream | |
hof.update(deme) | |
db.append(chain(*demes), gen=gen) | |
print db.stream | |
if gen % MIG_RATE == 0: | |
toolbox.migrate(demes) | |
gen += 1 | |
return demes, db, hof | |
if __name__ == "__main__": | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment