Skip to content

Instantly share code, notes, and snippets.

@fmder
Forked from cmd-ntrf/datalogger.py
Last active December 17, 2015 02:08
Show Gist options
  • Save fmder/5533193 to your computer and use it in GitHub Desktop.
Save fmder/5533193 to your computer and use it in GitHub Desktop.
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):
class Entry(OrderedDict):
db = self
def __str__(self):
return "\t".join(map(str, (self[name] for name in self.db.header)))
self.key = key
self.functions = {}
self.axis = 0
self.header = None
self.Entry = Entry
super(self.__class__, self).__init__()
def __getstate__(self):
state = {}
state['content'] = [OrderedDict(entry) for entry in self]
state['axis'] = self.axis
state['functions'] = self.functions
state['header'] = self.header
state['key'] = self.key
return state
def __setstate__(self, state):
self.extend(self.Entry(entry) for entry in state['content'])
self.axis = state['axis']
self.functions = state['functions']
self.header = state['header']
self.key = state['key']
def __reduce__(self):
return (self.__class__, (), self.__getstate__())
@property
def stream(self):
"""Retreive the formated unstreamed entries of the database including
the headers."""
startindex, self.buffindex = self.buffindex, len(self)
return self.__str__(startindex)
def register(self, name, function):
self.functions[name] = function
def select(self, *names):
if len(names) == 1:
return [entry[names[0]] for entry in self]
result = {}
for name in names:
result[name] = [entry[name] for entry in self]
return result
def append(self, data=[], **kargs):
if not self.header:
self.header = kargs.keys() + self.functions.keys()
values = numpy.array([self.key(elem) for elem in data])
entry = self.Entry.fromkeys(self.header, "")
for key, func in self.functions.iteritems():
entry[key] = func(values, self.axis)
entry.update(kargs)
list.append(self, entry)
def __str__(self, buffindex=-1):
if buffindex < 0:
startindex = 0
else:
startindex = buffindex
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))
if buffindex < 0:
text = [template.format(*self.header)]
else:
text = []
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 dbstream
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