source: framspy/evolalg/examples/multicriteria.py @ 1167

#TODO hof should be the complete non-dominated set from the entire process of evolution (all evaluated individuals), not limited in any way (remove '-hof_size'). Now we are not storing the entire Pareto front, but individuals that were better than others in hof [better=on all criteria?] at the time of inserting to hof. Hof has a size limit.
#TODO when -dissim is used, print its statistics just like all other criteria
#TODO in statistics, do not print "gen" (generation number) for each criterion
#TODO if possible, when saving the final .gen file, instead of fitness as a python tuple, save individual criteria - so instead of fitness:(0.005251036058321138, 0.025849976588613266), write "velocity:...\nvertpos:..." (this also applies to other .py examples in this directory)


import argparse
import logging
import os
import pickle
import sys
import copy
from enum import Enum

import numpy as np

from deap import base
from deap import tools

from FramsticksLib import FramsticksLib
from evolalg.base.lambda_step import LambdaStep
from evolalg.base.step import Step
from evolalg.dissimilarity.frams_dissimilarity import FramsDissimilarity
from evolalg.dissimilarity.levenshtein import LevenshteinDissimilarity
from evolalg.experiment import Experiment
from evolalg.fitness.fitness_step import FitnessStep
from evolalg.mutation_cross.frams_cross_and_mutate import FramsCrossAndMutate
from evolalg.population.frams_population import FramsPopulation
from evolalg.repair.remove.field import FieldRemove
from evolalg.repair.remove.remove import Remove
from evolalg.selection.nsga2 import NSGA2Selection
from evolalg.statistics.halloffame_stats import HallOfFameStatistics
from evolalg.statistics.multistatistics_deap import MultiStatistics
from evolalg.statistics.statistics_deap import StatisticsDeap
from evolalg.base.union_step import UnionStep
from evolalg.utils.population_save import PopulationSave


def ensureDir(string):
    if os.path.isdir(string):
        return string
    else:
        raise NotADirectoryError(string)


class Dissim(Enum):
    levenshtein = "levenshtein"
    frams = "frams"

    def __str__(self):
        return self.name



def parseArguments():
    parser = argparse.ArgumentParser(
        description='Run this program with "python -u %s" if you want to disable buffering of its output.' % sys.argv[
            0])
    parser.add_argument('-path', type=ensureDir, required=True,
                        help='Path to the Framsticks library without trailing slash.')
    parser.add_argument('-opt', required=True,
                        help='optimization criteria seperated with a comma: vertpos, velocity, distance, vertvel, lifespan, numjoints, numparts, numneurons, numconnections (and others as long as they are provided by the .sim file and its .expdef). Single or multiple criteria.')
    parser.add_argument('-lib', required=False, help="Filename of .so or .dll with the Framsticks library")

    parser.add_argument('-genformat', required=False, default="1",
                        help='Genetic format for the demo run, for example 4, 9, or B. If not given, f1 is assumed.')
    parser.add_argument('-sim', required=False, default="eval-allcriteria.sim",
                        help="Name of the .sim file with all parameter values. If you want to provide more files, separate them with a semicolon ';'.")
    parser.add_argument('-dissim', required=False, type=Dissim, default=Dissim.frams,
                        help='Dissimilarity measure, default: frams', choices=list(Dissim))
    parser.add_argument('-popsize', type=int, default=40, help="Population size (must be a multiple of 4), default: 40.") # mod 4 because of DEAP
    parser.add_argument('-generations', type=int, default=5, help="Number of generations, default: 5.")

    parser.add_argument('-max_numparts', type=int, default=None, help="Maximum number of Parts. Default: no limit")
    parser.add_argument('-max_numjoints', type=int, default=None, help="Maximum number of Joints. Default: no limit")
    parser.add_argument('-max_numneurons', type=int, default=None, help="Maximum number of Neurons. Default: no limit")
    parser.add_argument('-max_numconnections', type=int, default=None,
                        help="Maximum number of Neural connections. Default: no limit")

    parser.add_argument('-hof_size', type=int, default=10, help="Number of genotypes in Hall of Fame. Default: 10.")
    parser.add_argument('-hof_evaluations', type=int, default=20,
                        help="Number of final evaluations of each genotype in Hall of Fame to obtain reliable (averaged) fitness. Default: 20.")
    parser.add_argument('-checkpoint_path', required=False, default=None, help="Path to the checkpoint file")
    parser.add_argument('-checkpoint_interval', required=False, type=int, default=100, help="Checkpoint interval")
    parser.add_argument('-debug', dest='debug', action='store_true', help="Prints names of steps as they are executed")
    parser.set_defaults(debug=False)
    return parser.parse_args()


class NumPartsHigher(Remove):
    def __init__(self, max_number):
        super(NumPartsHigher, self).__init__()
        self.max_number = max_number

    def remove(self, individual):
        return individual.numparts > self.max_number


class NumJointsHigher(Remove):
    def __init__(self, max_number):
        super(NumJointsHigher, self).__init__()
        self.max_number = max_number

    def remove(self, individual):
        return individual.numjoints > self.max_number


class NumNeuronsHigher(Remove):
    def __init__(self, max_number):
        super(NumNeuronsHigher, self).__init__()
        self.max_number = max_number

    def remove(self, individual):
        return individual.numneurons > self.max_number


class NumConnectionsHigher(Remove):
    def __init__(self, max_number):
        super(NumConnectionsHigher, self).__init__()
        self.max_number = max_number

    def remove(self, individual):
        return individual.numconnections > self.max_number


class ReplaceWithHallOfFame(Step):
    def __init__(self, hof, *args, **kwargs):
        super(ReplaceWithHallOfFame, self).__init__(*args, **kwargs)
        self.hof = hof

    def call(self, population, *args, **kwargs):
        super(ReplaceWithHallOfFame, self).call(population)
        return list(self.hof.halloffame)


class DeapFitness(base.Fitness):
    weights = (1, 1)

    def __init__(self, *args, **kwargs):
        super(DeapFitness, self).__init__(*args, **kwargs)


class Nsga2Fitness:
    def __init__(self, fields):
        self.fields = fields
    def __call__(self, ind):
        setattr(ind, "fitness", DeapFitness(tuple(getattr(ind, _) for _ in self.fields)))



class ExtractField:
    def __init__(self, field_name):
        self.field_name = field_name
    def __call__(self, ind):
        return getattr(ind, self.field_name)

def extract_fitness(ind):
    return ind.fitness_raw


def load_experiment(path):
    with open(path, "rb") as file:
        experiment = pickle.load(file)
    print("Loaded experiment. Generation:", experiment.generation)
    return experiment


def create_experiment():
    parsed_args = parseArguments()
    frams_lib = FramsticksLib(parsed_args.path, parsed_args.lib, parsed_args.sim.split(";"))

    opt_dissim = []
    opt_fitness = []
    for crit in parsed_args.opt.split(','):
        try:
            Dissim(crit)
            opt_dissim.append(crit)
        except ValueError:
            opt_fitness.append(crit)
    if len(opt_dissim) > 1:
        raise ValueError("Only one type of dissimilarity supported")

    # Steps for generating first population
    init_stages = [
        FramsPopulation(frams_lib, parsed_args.genformat, parsed_args.popsize)
    ]

    # Selection procedure
    selection = NSGA2Selection(copy=True)

    # Procedure for generating new population. This steps will be run as long there is less than
    # popsize individuals in the new population
    new_generation_stages = [FramsCrossAndMutate(frams_lib, cross_prob=0.2, mutate_prob=0.9)]

    # Steps after new population is created. Executed exactly once per generation.
    generation_modifications = []

    # -------------------------------------------------
    # Fitness

    fitness_raw = FitnessStep(frams_lib, fields={**{_:_ for _ in opt_fitness},
                                                 "numparts": "numparts",
                                                 "numjoints": "numjoints",
                                                 "numneurons": "numneurons",
                                                 "numconnections": "numconnections"},
                              fields_defaults={parsed_args.opt: None, "numparts": float("inf"),
                                               "numjoints": float("inf"), "numneurons": float("inf"),
                                               "numconnections": float("inf"),
                                               **{_:None for _ in opt_fitness}
                                               },
                              evaluation_count=1)

    fitness_end = FitnessStep(frams_lib, fields={_:_ for _ in opt_fitness},
                              fields_defaults={parsed_args.opt: None},
                              evaluation_count=parsed_args.hof_evaluations)
    # Remove
    remove = []
    remove.append(FieldRemove(opt_fitness[0], None))  # Remove individuals if they have default value for fitness
    if parsed_args.max_numparts is not None:
        # This could be also implemented by "LambdaRemove(lambda x: x.numparts > parsed_args.num_parts)"
        # But this would not serialize in checkpoint.
        remove.append(NumPartsHigher(parsed_args.max_numparts))
    if parsed_args.max_numjoints is not None:
        remove.append(NumJointsHigher(parsed_args.max_numjoints))
    if parsed_args.max_numneurons is not None:
        remove.append(NumNeuronsHigher(parsed_args.max_numneurons))
    if parsed_args.max_numconnections is not None:
        remove.append(NumConnectionsHigher(parsed_args.max_numconnections))

    remove_step = UnionStep(remove)

    fitness_remove = UnionStep([fitness_raw, remove_step])

    init_stages.append(fitness_remove)
    new_generation_stages.append(fitness_remove)

    # -------------------------------------------------
    # Dissimilarity as one of the criteria
    dissim = None
    if len(opt_dissim) > 0 and Dissim(opt_dissim[0]) == Dissim.levenshtein:
        dissim = LevenshteinDissimilarity(reduction="mean", output_field="dissim")
    elif len(opt_dissim) > 0 and Dissim(opt_dissim[0]) == Dissim.frams:
        dissim = FramsDissimilarity(frams_lib, reduction="mean", output_field="dissim")

    if dissim is not None:
        init_stages.append(dissim)
        generation_modifications.append(dissim)

    if dissim is not None:
        nsga2_fittnes = Nsga2Fitness(["dissim"]+ opt_fitness)
    else:
        nsga2_fittnes = Nsga2Fitness(opt_fitness)

    init_stages.append(LambdaStep(nsga2_fittnes))
    generation_modifications.append(LambdaStep(nsga2_fittnes))

    # -------------------------------------------------
    # Statistics
    hall_of_fame = HallOfFameStatistics(parsed_args.hof_size, "fitness")  # Wrapper for halloffamae
    replace_with_hof = ReplaceWithHallOfFame(hall_of_fame)
    statistics_deap = MultiStatistics({fit:StatisticsDeap([
        ("avg", np.mean),
        ("stddev", np.std),
        ("min", np.min),
        ("max", np.max)
    ], ExtractField(fit)) for fit in opt_fitness}) # Wrapper for deap statistics

    statistics_union = UnionStep(
        [hall_of_fame,
        statistics_deap]
    )  # Union of two statistics steps.

    init_stages.append(statistics_union)
    generation_modifications.append(statistics_union)

    # -------------------------------------------------
    # End stages: this will execute exactly once after all generations.
    end_stages = [
        replace_with_hof,
        fitness_end,
        PopulationSave("halloffame.gen", provider=hall_of_fame.halloffame, fields={"genotype": "genotype",
                                                                                   "fitness": "fitness"})]
    # ...but custom fields can be added, e.g. "custom": "recording"

    # -------------------------------------------------

    # Experiment creation

    experiment = Experiment(init_population=init_stages,
                            selection=selection,
                            new_generation_steps=new_generation_stages,
                            generation_modification=generation_modifications,
                            end_steps=end_stages,
                            population_size=parsed_args.popsize,
                            checkpoint_path=parsed_args.checkpoint_path,
                            checkpoint_interval=parsed_args.checkpoint_interval
                            )
    return experiment


def main():
    print("Running experiment with", sys.argv)
    parsed_args = parseArguments()
    
    if parsed_args.popsize % 4 != 0:
        raise ValueError("popsize must be a multiple of 4 (for example %d)." % (parsed_args.popsize//4*4)) # required by deap.tools.selTournamentDCD() 
        
    if parsed_args.debug:
        logging.basicConfig(level=logging.DEBUG)

    if parsed_args.checkpoint_path is not None and os.path.exists(parsed_args.checkpoint_path):
        experiment = load_experiment(parsed_args.checkpoint_path)
    else:
        experiment = create_experiment()
        experiment.init()  # init is mandatory

    experiment.run(parsed_args.generations)

    # Next call for experiment.run(10) will do nothing. Parameter 10 specifies how many generations should be
    # in one experiment. Previous call generated 10 generations.
    # Example 1:
    # experiment.init()
    # experiment.run(10)
    # experiment.run(12)
    # #This will run for total of 12 generations
    #
    # Example 2
    # experiment.init()
    # experiment.run(10)
    # experiment.init()
    # experiment.run(10)
    # # All work produced by first run will be "destroyed" by second init().


if __name__ == '__main__':
    main()