"""PINSGA2 — interactive NSGA-II that guides selection with a value function learned from decision-maker preferences."""
import sys
from abc import ABC, abstractmethod
import numpy as np
from pymoo.algorithms.base.genetic import GeneticAlgorithm
from pymoo.algorithms.moo.nsga2 import binary_tournament
from pymoo.docs import parse_doc_string
from pymoo.operators.crossover.sbx import SBX
from pymoo.operators.mutation.pm import PM
from pymoo.operators.sampling.rnd import FloatRandomSampling
from pymoo.operators.selection.tournament import TournamentSelection
from pymoo.operators.survival.rank_and_crowding import RankAndCrowding
from pymoo.termination.default import DefaultMultiObjectiveTermination
from pymoo.util import value_functions as mvf
from pymoo.util.display.multi import MultiObjectiveOutput
from pymoo.util.nds.non_dominated_sorting import NonDominatedSorting
from pymoo.util.reference_direction import select_points_with_maximum_distance
from pymoo.util.vf_dominator import VFDominator
# =========================================================================================================
# Implementation
# =========================================================================================================
class AutomatedDM(ABC):
def __init__(self, get_pairwise_ranks_func=None):
self.get_pairwise_ranks_func = get_pairwise_ranks_func
@abstractmethod
def makeDecision(self, F):
pass
def makePairwiseDecision(self, F):
def dm(F):
return self.makeDecision(F)
ranks = self.get_pairwise_ranks_func(F, 1, dm=dm)
return ranks
[docs]
class PINSGA2(GeneticAlgorithm):
def __init__(
self,
pop_size=100,
sampling=FloatRandomSampling(),
selection=TournamentSelection(func_comp=binary_tournament),
crossover=SBX(eta=15, prob=0.9),
mutation=PM(eta=20),
output=MultiObjectiveOutput(),
tau=10,
eta=4,
opt_method="trust-constr",
vf_type="poly",
eps_max=1000,
vf_max_iter=400,
ranking_type="pairwise",
presi_signs=None,
automated_dm=None,
verbose=False,
**kwargs,
):
self.survival = RankAndCrowding(
nds=NonDominatedSorting(dominator=VFDominator(self))
)
super().__init__(
pop_size=pop_size,
sampling=sampling,
selection=selection,
crossover=crossover,
mutation=mutation,
survival=self.survival,
output=output,
advance_after_initial_infill=True,
**kwargs,
)
self.termination = DefaultMultiObjectiveTermination()
self.tournament_type = "comp_by_dom_and_crowding"
self.ranking_type = ranking_type
self.presi_signs = presi_signs
self.vf_type = vf_type
self.opt_method = opt_method
self.tau = tau
self.eta = eta
self.eta_F = []
self.vf_res = None
self.v2 = None
self.vf_plot_flag = False
self.vf_plot = None
self.historical_F = None
self.prev_pop = None
self.fronts = []
self.eps_max = eps_max
# Cap the scipy VF fit; infeasible instances otherwise grind to scipy's
# maxiter=1000 and are discarded anyway. Set None to restore the full budget.
self.vf_max_iter = vf_max_iter
self.verbose = verbose
if automated_dm is not None:
automated_dm.get_pairwise_ranks_func = self._get_pairwise_ranks
self.automated_dm = automated_dm
def _warn(self, msg):
if self.verbose:
sys.stderr.write(msg)
@staticmethod
def _prompt_for_ranks(F, presi_signs):
for e, f in enumerate(F):
print("Solution %d %s" % (e + 1, f * presi_signs))
dim = F.shape[0]
raw_ranks = input(
f'Ranks (e.g., "3, {dim}, ..., 1" for 3rd best, {dim}th best, ..., 1st best): '
)
if raw_ranks == "":
ranks = []
else:
ranks = [int(raw_rank) for raw_rank in raw_ranks.split()]
return ranks
@staticmethod
def _present_ranks(F, dm_ranks, presi_signs):
print("Solutions are ranked as:")
for e, f in enumerate(F):
print("Solution %d %s: Rank %d" % (e + 1, f * presi_signs, dm_ranks[e]))
@staticmethod
def _get_pairwise_ranks(F, presi_signs, dm=None):
if not dm:
def dm(F):
return input(
"\nWhich solution do you like best?\n"
f"[a] {F[0]}\n"
f"[b] {F[1]}\n"
"[c] These solutions are equivalent.\n--> "
)
# initialize empty ranking
_ranks = []
for i, f in enumerate(F):
# handle empty case, put first element in first place
if not _ranks:
_ranks.append([i])
else:
inserted = False
# for each remaining elements, compare to all currently ranked elements
for j, group in enumerate(_ranks):
# get pairwise preference from user
while True:
points_to_compare = np.array(
[f * presi_signs, F[group[0]] * presi_signs]
)
preference_raw = dm(points_to_compare)
preference = preference_raw.strip().lower()
if preference in ["a", "b", "c"]:
break
print("Invalid input. Please enter 'a', 'b', or 'c'.")
# if better than currently ranked element place before that element
if preference == "a":
_ranks.insert(j, [i])
inserted = True
break
# if equal to currently ranked element place with that element
elif preference == "c":
group.append(i)
inserted = True
break
# if found to be worse than all place at the end
if not inserted:
_ranks.append([i])
ranks = np.zeros(len(F), dtype=int)
for rank, group in enumerate(_ranks):
for index in group:
ranks[index] = rank
return np.array(ranks) + 1
@staticmethod
def _get_ranks(F, presi_signs):
ranks_invalid = True
dim = F.shape[0]
print(
f"Give each solution a ranking, with 1 being the highest score, and {dim} being the lowest score:"
)
ranks = PINSGA2._prompt_for_ranks(F, presi_signs)
while ranks_invalid:
fc = F.shape[0]
if len(ranks) > 0 and max(ranks) <= fc and min(ranks) >= 1:
ranks_invalid = False
else:
print("Invalid ranks given. Please try again")
ranks = PINSGA2._prompt_for_ranks(F, presi_signs)
return np.array(ranks)
def _reset_dm_preference(self):
self._warn("Back-tracking and removing DM preference from search.")
self.eta_F = []
self.vf_res = None
self.v2 = None
self.vf_plot_flag = False
self.vf_plot = None
self.pop = self.prev_pop
def _advance(self, infills=None, **kwargs):
super()._advance(infills=infills, **kwargs)
rank, F = self.pop.get("rank", "F")
self.fronts = rank
F = F[rank == 0]
if self.historical_F is not None:
self.historical_F = np.vstack((self.historical_F, F))
else:
self.historical_F = F
to_find = self.eta if F.shape[0] >= self.eta else F.shape[0]
if self.presi_signs is None:
self.presi_signs = np.ones(F.shape[1])
# Eta is the number of solutions displayed to the DM
eta_F_indices = select_points_with_maximum_distance(
F, to_find, random_state=self.random_state
)
self.eta_F = F[eta_F_indices]
self.eta_F = self.eta_F[self.eta_F[:, 0].argsort()]
# Remove duplicate rows
self.eta_F = np.unique(self.eta_F, axis=0)
# A frozen view of the optimization each tau generations
self.paused_F = F
# Record the previous population in case we need to back track
self.prev_pop = self.pop
dm_time = self.n_gen % self.tau == 0
# Check whether we have more than one solution
if dm_time and len(self.eta_F) < 2:
self._warn("Population only contains one non-dominated solution. ")
self._reset_dm_preference()
elif dm_time:
# Check if the DM is a machine or a human
if self.automated_dm is None:
# Human DM
if self.ranking_type == "absolute":
dm_ranks = PINSGA2._get_ranks(self.eta_F, self.presi_signs)
elif self.ranking_type == "pairwise":
dm_ranks = PINSGA2._get_pairwise_ranks(self.eta_F, self.presi_signs)
PINSGA2._present_ranks(self.eta_F, dm_ranks, self.presi_signs)
else:
raise ValueError(
"Invalid ranking type [%s] given." % self.ranking_type
)
else:
# Automated DM
if self.ranking_type == "absolute":
dm_ranks = self.automated_dm.makeDecision(self.eta_F)
elif self.ranking_type == "pairwise":
dm_ranks = self.automated_dm.makePairwiseDecision(self.eta_F)
else:
raise ValueError(
"Invalid ranking type [%s] given." % self.ranking_type
)
if len(set(rank)) == 0:
self._warn("No preference between any two points provided.")
self._reset_dm_preference()
return
eta_F = self.eta_F
while eta_F.shape[0] > 1:
if self.vf_type == "linear":
vf_res = mvf.create_linear_vf(
eta_F * -1,
dm_ranks.tolist(),
eps_max=self.eps_max,
method=self.opt_method,
max_iter=self.vf_max_iter,
)
elif self.vf_type == "poly":
vf_res = mvf.create_poly_vf(
eta_F * -1,
dm_ranks.tolist(),
eps_max=self.eps_max,
method=self.opt_method,
max_iter=self.vf_max_iter,
)
else:
raise ValueError("Value function %s not supported" % self.vf_type)
# check if we were able to model the VF
if vf_res.fit:
self.vf_res = vf_res
self.vf_plot_flag = True
self.v2 = self.vf_res.vf(eta_F[dm_ranks[1] - 1] * -1).item()
break
else:
# If we didn't the model, try to remove the least preferred point and try to refit
self._warn("Could not fit a function to the DM preference")
if eta_F.shape[0] == 2:
# If not, reset and use normal domination
self._warn("Removing DM preference")
self._reset_dm_preference()
break
else:
self._warn(
"Removing the second best preferred solution from the fit."
)
# ranks start at 1, not zero
rank_to_remove = dm_ranks[1]
eta_F = np.delete(eta_F, rank_to_remove - 1, axis=0)
dm_ranks = np.concatenate(([dm_ranks[0]], dm_ranks[2:]))
# update the ranks, since we just removed one
dm_ranks[dm_ranks > rank_to_remove] = (
dm_ranks[dm_ranks > rank_to_remove] - 1
)
parse_doc_string(PINSGA2.__init__)
