"""
Handles creation of genomes, either from scratch or by sexual or
asexual reproduction from parents.
"""
from __future__ import division, print_function
import math
import random
import warnings
from itertools import count
from sys import stderr, float_info
from neat.config import ConfigParameter, DefaultClassConfig
from neat.math_util import mean, NORM_EPSILON
from neat.six_util import iteritems, itervalues
# TODO: Provide some sort of optional cross-species performance criteria, which
# are then used to control stagnation and possibly the mutation rate
# configuration. This scheme should be adaptive so that species do not evolve
# to become "cautious" and only make very slow progress.
[docs]class DefaultReproduction(DefaultClassConfig):
"""
Implements the default NEAT-python reproduction scheme:
explicit fitness sharing with fixed-time species stagnation.
"""
[docs] @classmethod
def parse_config(cls, param_dict):
return DefaultClassConfig(param_dict,
[ConfigParameter('elitism', int, 0),
ConfigParameter('survival_threshold', float, 0.2),
ConfigParameter('min_species_size', int, 2),
ConfigParameter('fitness_min_divisor', float, 1.0),
ConfigParameter('double_mutate_clones', bool, False)])
def __init__(self, config, reporters, stagnation):
# pylint: disable=super-init-not-called
self.reproduction_config = config
self.genome_config = None
self.reporters = reporters
self.genome_indexer = count(1)
self.stagnation = stagnation
self.ancestors = {}
self.fitness_min_divisor = config.fitness_min_divisor
if config.survival_threshold <= 0.0: # NEEDS TEST
raise ValueError(
"Survival_threshold cannot be 0 or negative ({0:n})".format(
config.survival_threshold))
if config.survival_threshold > 1.0: # NEEDS TEST
raise ValueError(
"Survival_threshold cannot be above 1.0 ({0:n})".format(
config.survival_threshold))
if config.survival_threshold < NORM_EPSILON: # NEEDS TEST!
print("Survival_threshold {0:n} is too low; increasing to {1:n}".format(
config.survival_threshold, NORM_EPSILON),
file=stderr)
stderr.flush()
config.survival_threshold = NORM_EPSILON
if config.fitness_min_divisor < 0.0: # NEEDS TEST
raise ValueError(
"Fitness_min_divisor cannot be negative ({0:n})".format(
config.fitness_min_divisor))
elif config.fitness_min_divisor == 0.0:
self.fitness_min_divisor = NORM_EPSILON
elif config.fitness_min_divisor < float_info.epsilon:
print("Fitness_min_divisor {0:n} is too low; increasing to {1:n}".format(
config.fitness_min_divisor,float_info.epsilon), file=stderr)
stderr.flush()
config.fitness_min_divisor = float_info.epsilon
self.fitness_min_divisor = float_info.epsilon
if config.min_species_size < 2: # NEEDS TEST
raise ValueError(
"Min_species_size must be at least 2 (not {0:n}) for crossover parents".format(
config.min_species_size))
[docs] def create_new(self, genome_type, genome_config, num_genomes):
self.genome_config = genome_config # for get_species_size_info
new_genomes = {}
for dummy in range(num_genomes):
key = next(self.genome_indexer)
g = genome_type(key)
g.configure_new(genome_config)
new_genomes[key] = g
self.ancestors[key] = tuple()
return new_genomes
def get_species_size_info(self): # DOCUMENT!
to_return_dict = {}
to_return_dict['min_size'] = max(self.reproduction_config.elitism,
self.reproduction_config.min_species_size)
num_try = max(2,to_return_dict['min_size'])
to_return_dict['min_OK_size'] = to_return_dict['min_good_size'] = int(
math.ceil(num_try/self.reproduction_config.survival_threshold))
while ((num_try*(num_try-1)/2) # number of possible parent combinations
< (to_return_dict['min_good_size']-num_try)) and (to_return_dict['min_good_size']
< 50): # half of min pop size 100
num_try += 1
to_return_dict['min_good_size'] = int(
math.ceil(num_try/self.reproduction_config.survival_threshold))
# below - for info about weight, disjoint coefficients
to_return_dict['genome_config'] = self.genome_config
# for max_stagnation
to_return_dict['stagnation'] = self.stagnation
return to_return_dict
[docs] @staticmethod
def compute_spawn(adjusted_fitness, previous_sizes, pop_size, min_species_size):
"""Compute the proper number of offspring per species (proportional to fitness)."""
af_sum = sum(adjusted_fitness)
spawn_amounts = []
for af, ps in zip(adjusted_fitness, previous_sizes):
if af_sum > 0:
s = max(min_species_size, af / af_sum * pop_size)
else:
s = min_species_size
d = (s - ps) * 0.5
c = int(round(d))
spawn = ps
if abs(c) > 0:
spawn += c
elif d > 0:
spawn += 1
elif d < 0:
spawn -= 1
spawn_amounts.append(spawn)
# Normalize the spawn amounts so that the next generation is roughly
# the population size requested by the user.
total_spawn = sum(spawn_amounts)
norm = pop_size / total_spawn
spawn_amounts = [max(min_species_size, int(round(n * norm))) for n in spawn_amounts]
return spawn_amounts
[docs] def reproduce(self, config, species, pop_size, generation):
"""
Handles creation of genomes, either from scratch or by sexual or
asexual reproduction from parents.
"""
# TODO: I don't like this modification of the species and stagnation objects,
# because it requires internal knowledge of the objects.
if pop_size < (2*max(self.reproduction_config.elitism,
self.reproduction_config.min_species_size)):
raise ValueError(
"Population size must be at least {0:n}, not {1:n}".format(
(2*max(self.reproduction_config.elitism,
self.reproduction_config.min_species_size)),
pop_size))
if (generation == 0) and (pop_size
< (2*math.ceil(2/self.reproduction_config.survival_threshold))):
warnings.warn("Population size {0:n} is too small".format(pop_size) # NEEDS TEST!
+ " - with a survival_threshold of {0:n},".format(
self.reproduction_config.survival_threshold)
+ " a minimum of {0:n} is recommended".format(
2*math.ceil(2/self.reproduction_config.survival_threshold)))
# Filter out stagnated species, collect the set of non-stagnated
# species members, and compute their average adjusted fitness.
# The average adjusted fitness scheme (normalized to the interval
# [0, 1]) allows the use of negative fitness values without
# interfering with the shared fitness scheme.
all_fitnesses = []
remaining_species = []
for stag_sid, stag_s, stagnant in self.stagnation.update(species, generation):
if stagnant:
self.reporters.species_stagnant(stag_sid, stag_s)
del(species.species[stag_sid]) # for non-ref-tracking garbage collection
else:
all_fitnesses.extend(stag_s.get_fitnesses())
remaining_species.append(stag_s)
# The above comment was not quite what was happening - now getting fitnesses
# only from members of non-stagnated species.
# No species left.
if not remaining_species:
species.species = {}
return {} # was []
# Find minimum/maximum fitness across the entire population, for use in
# species adjusted fitness computation.
min_fitness = min(all_fitnesses)
max_fitness = max(all_fitnesses)
# Do not allow the fitness range to be zero, as we divide by it below.
fitness_range = max(self.fitness_min_divisor, max_fitness - min_fitness)
for afs in remaining_species:
# Compute adjusted fitness.
# TODO: One variant suggested by Stanley on his webpage is to only use the
# fitnesses of the members that will be passed on to the next generation and/or
# get a chance to reproduce. This is a compromise between mean and max.
msf = mean([m.fitness for m in itervalues(afs.members)])
af = (msf - min_fitness) / fitness_range
afs.reproduction_namespace.adjusted_fitness = af
adjusted_fitnesses = [s.reproduction_namespace.adjusted_fitness for s in remaining_species]
avg_adjusted_fitness = mean(adjusted_fitnesses) # type: float
self.reporters.info("Average adjusted fitness: {:.3f}".format(avg_adjusted_fitness))
# Compute the number of new members for each species in the new generation.
previous_sizes = [len(s.members) for s in remaining_species]
min_species_size = self.reproduction_config.min_species_size
# Isn't the effective min_species_size going to be max(min_species_size,
# self.reproduction_config.elitism)? That would probably produce more accurate tracking
# of population sizes and relative fitnesses... doing; documented.
min_species_size = max(min_species_size,self.reproduction_config.elitism)
spawn_amounts = self.compute_spawn(adjusted_fitnesses, previous_sizes,
pop_size, min_species_size)
new_population = {}
species.species = {}
for spawn, s in zip(spawn_amounts, remaining_species):
# If elitism is enabled, each species always at least gets to retain its elites.
spawn = max(spawn, self.reproduction_config.elitism)
assert spawn > 0
# The species has at least one member for the next generation, so retain it.
old_members = list(iteritems(s.members))
s.members = {}
species.species[s.key] = s
# Sort members in order of descending fitness.
old_members.sort(reverse=True, key=lambda x: x[1].fitness)
# Transfer elites to new generation.
if self.reproduction_config.elitism > 0:
for i, m in old_members[:self.reproduction_config.elitism]:
new_population[i] = m
spawn -= 1
if spawn <= 0:
continue
# Only use the survival threshold fraction to use as parents for the next generation.
repro_cutoff = int(math.ceil(self.reproduction_config.survival_threshold *
len(old_members)))
# Use at least two parents no matter what the threshold fraction result is.
repro_cutoff = max(repro_cutoff, 2)
old_members = old_members[:repro_cutoff]
# Randomly choose parents and produce the number of offspring allotted to the species.
while spawn > 0:
spawn -= 1
parent1_id, parent1 = random.choice(old_members)
parent2_id, parent2 = random.choice(old_members)
# Note that if the parents are not distinct,
# crossover (before mutation) will produce a
# genetically identical clone of the parent (but with a different ID).
gid = next(self.genome_indexer)
child = config.genome_type(gid)
child.configure_crossover(parent1, parent2, config.genome_config)
child.mutate(config.genome_config)
if self.reproduction_config.double_mutate_clones and (parent1_id == parent2_id):
child.mutate(config.genome_config)
new_population[gid] = child
self.ancestors[gid] = (parent1_id, parent2_id)
return new_population