Source code for edo.fitness

""" Functions for calculating individual and population fitness. """

from pathlib import Path

import dask
import pandas as pd


@dask.delayed
def get_fitness(individual, fitness, **kwargs):
    """ Return the fitness score of the individual. """

    if individual.fitness is None:
        individual.fitness = fitness(individual, **kwargs)

    return individual.fitness


[docs]def get_population_fitness(population, fitness, processes=None, **kwargs): """Return the fitness of each individual in the population. This can be done in parallel by specifying a number of cores to use for independent processes.""" tasks = ( get_fitness(individual, fitness, **kwargs) for individual in population ) if processes is None: out = dask.compute(*tasks, scheduler="single-threaded") else: out = dask.compute(*tasks, num_workers=processes) return list(out)
[docs]def write_fitness(fitness, generation, root): """ Write the generation fitness to file in the ``root`` directory. """ path = Path(root) path.mkdir(parents=True, exist_ok=True) size = len(fitness) if generation == 0: pd.DataFrame( { "fitness": fitness, "generation": generation, "individual": range(size), } ).to_csv(path / "fitness.csv", index=False) else: with open(path / "fitness.csv", "a") as fit_file: for fit, gen, ind in zip(fitness, [generation] * size, range(size)): fit_file.write(f"{fit},{gen},{ind}\n")