MH-P2/src/genetic_algorithm.py

from numpy import sum, append, arange, delete, where
from numpy.random import randint, choice, shuffle
from pandas import DataFrame
from math import ceil


def get_row_distance(source, destination, data):
    row = data.query(
        """(source == @source and destination == @destination) or \
        (source == @destination and destination == @source)"""
    )
    return row["distance"].values[0]


def compute_distance(element, individual, data):
    accumulator = 0
    distinct_elements = individual.query(f"point != {element}")
    for _, item in distinct_elements.iterrows():
        accumulator += get_row_distance(
            source=element, destination=item.point, data=data
        )
    return accumulator


def generate_individual(n, m, data):
    individual = DataFrame(columns=["point", "distance", "fitness"])
    individual["point"] = choice(n, size=m, replace=False)
    individual["distance"] = individual["point"].apply(
        func=compute_distance, individual=individual, data=data
    )
    return individual


def evaluate_individual(individual, data):
    fitness = []
    genotype = individual.point.values
    distances = data.query(f"source in @genotype and destination in @genotype")
    for item in genotype[:-1]:
        element_df = distances.query(f"source == {item} or destination == {item}")
        max_distance = element_df["distance"].astype(float).max()
        fitness = append(arr=fitness, values=max_distance)
        distances = distances.query(f"source != {item} and destination != {item}")
    return sum(fitness)


def select_distinct_genes(matching_genes, parents, m):
    cutoff = randint(m)
    distinct_indexes = delete(arange(m), matching_genes)
    first_parent_genes = parents[0].point.iloc[distinct_indexes[cutoff:]]
    second_parent_genes = parents[1].point.iloc[distinct_indexes[:cutoff]]
    return first_parent_genes, second_parent_genes


def select_random_genes(matching_genes, parents, m):
    random_parent = parents[randint(len(parents))]
    distinct_indexes = delete(arange(m), matching_genes)
    genes = random_parent.point.iloc[distinct_indexes].values
    shuffle(genes)
    return genes


def repair_offspring(offspring, parents, m):
    while len(offspring) != m:
        if len(offspring) > m:
            best_index = offspring["distance"].astype(float).idxmax()
            offspring.drop(index=best_index, inplace=True)
        elif len(offspring) < m:
            random_parent = parents[randint(len(parents))]
            best_index = random_parent["distance"].astype(float).idxmax()
            best_point = random_parent["point"].loc[best_index]
            offspring = offspring.append(
                {"point": best_point, "distance": 0}, ignore_index=True
            )
            random_parent.drop(index=best_index, inplace=True)
    return offspring


def get_matching_genes(parents):
    first_parent = parents[0].point
    second_parent = parents[1].point
    return where(first_parent == second_parent)


def populate_offspring(values):
    offspring = DataFrame(columns=["point", "distance", "fitness"])
    for element in values:
        aux = DataFrame(columns=["point", "distance", "fitness"])
        aux["point"] = element
        offspring = offspring.append(aux)
    offspring["distance"] = 0
    offspring["fitness"] = 0
    offspring = offspring[1:]
    return offspring


def uniform_crossover(parents, m):
    matching_indexes = get_matching_genes(parents)
    matching_genes = parents[0].point.iloc[matching_indexes]
    first_genes, second_genes = select_distinct_genes(matching_genes, parents, m)
    offspring = populate_offspring(values=[matching_genes, first_genes, second_genes])
    viable_offspring = repair_offspring(offspring, parents, m)
    return viable_offspring


def position_crossover(parents, m):
    matching_genes = get_matching_genes(parents)
    shuffled_genes = select_random_genes(matching_genes, parents, m)
    first_offspring = populate_offspring(values=[matching_genes, shuffled_genes])
    second_offspring = populate_offspring(values=[matching_genes, shuffled_genes])
    return [first_offspring, second_offspring]


def crossover(mode, parents, m):
    if mode == "uniform":
        return uniform_crossover(parents, m)
    return position_crossover(parents, m)


def element_in_dataframe(individual, element):
    duplicates = individual.query(f"point == {element}")
    return not duplicates.empty


def select_new_gene(individual, n):
    while True:
        new_gene = randint(n)
        if not element_in_dataframe(individual=individual, element=new_gene):
            return new_gene


def mutate(population, n, probability=0.001):
    expected_mutations = len(population) * n * probability
    individuals = []
    genes = []
    for _ in range(ceil(expected_mutations)):
        individuals.append(randint(n))
        current_individual = individuals[-1]
        genes.append(population[current_individual].sample().index)
    for ind, gen in zip(individuals, genes):
        individual = population[ind]
        individual["point"].iloc[gen] = select_new_gene(individual, n)
        individual["distance"].iloc[gen] = 0
    return population


def tournament_selection(population):
    individuals = population.sample(n=2)
    best_index = population["distance"].idxmax()
    return individuals.iloc[best_index]


def generational_replacement(previous_population, current_population):
    new_population = current_population
    best_previous_individual = max(previous_population, key=lambda x: x.fitness)
    if best_previous_individual not in new_population:
        worst_index = new_population.index(min(new_population, key=lambda x: x.fitness))
        new_population[worst_index] = best_previous_individual
    return new_population


def stationary_replacement(previous_population, current_population):
    new_population = previous_population
    return new_population


def replace_population(previous_population, current_population, mode):
    if mode == "generational":
        return generational_replacement(previous_population, current_population)
    return stationary_replacement(previous_population, current_population)


def genetic_algorithm(n, m, data, mode):
    population = [generate_individual(n, m, data) for _ in range(n)]