"""
Makes possible reporter classes,
which are triggered on particular events and may provide information to the user,
may do something else such as checkpointing, or may do both.
"""
from __future__ import division, print_function
import time
from neat.math_util import mean, stdev
from neat.six_util import itervalues, iterkeys
# TODO: Add a curses-based reporter.
[docs]class ReporterSet(object):
"""
Keeps track of the set of reporters
and gives methods to dispatch them at appropriate points.
"""
def __init__(self):
self.reporters = []
[docs] def add(self, reporter):
self.reporters.append(reporter)
[docs] def remove(self, reporter):
self.reporters.remove(reporter)
[docs] def start_generation(self, gen):
for r in self.reporters:
r.start_generation(gen)
[docs] def end_generation(self, config, population, species_set):
for r in self.reporters:
r.end_generation(config, population, species_set)
[docs] def post_evaluate(self, config, population, species, best_genome):
for r in self.reporters:
r.post_evaluate(config, population, species, best_genome)
[docs] def post_reproduction(self, config, population, species):
for r in self.reporters:
r.post_reproduction(config, population, species)
[docs] def complete_extinction(self):
for r in self.reporters:
r.complete_extinction()
[docs] def found_solution(self, config, generation, best):
for r in self.reporters:
r.found_solution(config, generation, best)
[docs] def species_stagnant(self, sid, species):
for r in self.reporters:
r.species_stagnant(sid, species)
[docs] def info(self, msg):
for r in self.reporters:
r.info(msg)
[docs]class BaseReporter(object):
"""Definition of the reporter interface expected by ReporterSet."""
[docs] def start_generation(self, generation):
pass
[docs] def end_generation(self, config, population, species_set):
pass
[docs] def post_evaluate(self, config, population, species, best_genome):
pass
[docs] def post_reproduction(self, config, population, species):
pass
[docs] def complete_extinction(self):
pass
[docs] def found_solution(self, config, generation, best):
pass
[docs] def species_stagnant(self, sid, species):
pass
[docs] def info(self, msg):
pass
[docs]class StdOutReporter(BaseReporter):
"""Uses `print` to output information about the run; an example reporter class."""
def __init__(self, show_species_detail):
self.show_species_detail = show_species_detail
self.generation = None
self.generation_start_time = None
self.generation_times = []
self.num_extinctions = 0
def start_generation(self, generation):
self.generation = generation
print('\n ****** Running generation {0} ****** \n'.format(generation))
self.generation_start_time = time.time()
def end_generation(self, config, population, species_set):
ng = len(population)
ns = len(species_set.species)
if self.show_species_detail:
print('Population of {0:d} members in {1:d} species:'.format(ng, ns))
sids = list(iterkeys(species_set.species))
sids.sort()
print(" ID age size fitness adj fit stag")
print(" ==== === ==== ======= ======= ====")
for sid in sids:
s = species_set.species[sid]
a = self.generation - s.created
n = len(s.members)
if s.stagnation_namespace.fitness is None:
f = "--"
else:
f = "{:.1f}".format(s.stagnation_namespace.fitness)
if s.reproduction_namespace.adjusted_fitness is None:
af = "--"
else:
af = "{:.3f}".format(s.reproduction_namespace.adjusted_fitness)
st = self.generation - s.stagnation_namespace.last_improved
print(
" {: >4} {: >3} {: >4} {: >7} {: >7} {: >4}".format(sid, a, n, f, af, st))
else:
print('Population of {0:d} members in {1:d} species'.format(ng, ns))
elapsed = time.time() - self.generation_start_time
self.generation_times.append(elapsed)
self.generation_times = self.generation_times[-10:]
average = sum(self.generation_times) / len(self.generation_times)
print('Total extinctions: {0:d}'.format(self.num_extinctions))
if len(self.generation_times) > 1:
print("Generation time: {0:.3f} sec ({1:.3f} average)".format(elapsed, average))
else:
print("Generation time: {0:.3f} sec".format(elapsed))
def post_evaluate(self, config, population, species, best_genome):
# pylint: disable=no-self-use
fitnesses = [c.fitness for c in itervalues(population)]
fit_mean = mean(fitnesses)
fit_std = stdev(fitnesses)
best_species_id = species.get_species_id(best_genome.key)
print('Population\'s average fitness: {0:3.5f} stdev: {1:3.5f}'.format(fit_mean, fit_std))
print(
'Best fitness: {0:3.5f} - size: {1!r} - species {2} - id {3}'.format(best_genome.fitness,
best_genome.size(),
best_species_id,
best_genome.key))
def complete_extinction(self):
self.num_extinctions += 1
print('All species extinct.')
def found_solution(self, config, generation, best):
print('\nBest individual in generation {0} meets fitness threshold - complexity: {1!r}'.format(
self.generation, best.size()))
def species_stagnant(self, sid, species):
if self.show_species_detail:
print("\nSpecies {0} with {1} members is stagnated: removing it".format(
sid, len(species.members)))
def info(self, msg):
print(msg)