Skip to content

Instantly share code, notes, and snippets.

@cmd-ntrf
Last active December 17, 2015 00:59
Show Gist options
  • Save cmd-ntrf/5525264 to your computer and use it in GitHub Desktop.
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.
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")
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()
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