Source code for zellij.strategies.tools.tree_search

# @Author: Thomas Firmin <ThomasFirmin>
# @Date:   2022-05-03T15:41:48+02:00
# @Email:  thomas.firmin@univ-lille.fr
# @Project: Zellij
# @Last modified by:   tfirmin
# @Last modified time: 2022-10-03T22:37:38+02:00
# @License: CeCILL-C (http://www.cecill.info/index.fr.html)

import numpy as np
import abc
import copy

from collections import defaultdict
from itertools import groupby

import logging

logger = logging.getLogger("zellij.tree_search")


























##########
# DIRECT #
##########


[docs]class Potentially_Optimal_Rectangle(Tree_search): """Potentially_Optimal_Rectangle Potentially Optimal Rectangle algorithm (POR), is a the selection strategy comming from DIRECT. Attributes ---------- open : list[Fractal] Initial Open list containing not explored nodes from the partition tree. max_depth : int maximum depth of the partition tree. Q : int, default=1 Q-Best_first_search, at each get_next, tries to return Q nodes. reverse : boolean, default=False if False do a descending sort the open list, else do an ascending sort Methods ------- add(self,c) Add a node c to the fractal tree get_next(self) Get the next node to evaluate See Also -------- Fractal : Abstract class defining what a fractal is. FDA : Fractal Decomposition Algorithm Tree_search : Base class Beam_search : Memory efficient tree search algorithm based on BestFS Cyclic_best_first_search : Hybrid between DFS and BestFS """ def __init__(self, open, max_depth=600, error=1e-4, maxdiv=3000): """__init__(self, open, max_depth, Q=1, reverse=False, error=1e-4) Parameters ---------- open : list[Fractal] Initial Open list containing not explored nodes from the partition tree. max_depth : int maximum depth of the partition tree. Q : int, default=1 Q-Best_first_search, at each get_next, tries to return Q nodes. reverse : boolean, default=False if False do a descending sort the open list, else do an ascending sort error : float, default=1e-4 Small value which determines when an evaluation should be considered as good as the best solution found so far. """ super().__init__(open, max_depth) ############## # PARAMETERS # ############## self.error = error self.maxdiv = maxdiv ############# # VARIABLES # ############# self.maxi1 = np.full(self.maxdiv, -float("inf"), dtype=float) self.mini2 = np.full(self.maxdiv, float("inf"), dtype=float) self.next_frontier = [] min = [c.score for c in self.open] self.best_score = np.min(min)
[docs] def add(self, c): self.next_frontier.append(c) self.best_score = c.loss.best_score
[docs] def get_next(self): if len(self.next_frontier) > 0: # sort potentially optimal rectangle by length (incresing) # then by score self.open += sorted( self.next_frontier, key=lambda x: (-x.length, x.score) ) # clip open list to maxdiv self.open = sorted(self.open, key=lambda x: (-x.length, x.score))[ : self.maxdiv ] self.next_frontier = [] if len(self.open): self.maxi1.fill(-float("inf")) self.mini2.fill(float("inf")) groups = groupby(self.open, lambda x: x.length) idx = self.optimal(groups) if idx: for i in reversed(idx): self.close.append(self.open.pop(i)) return True, self.close[-len(idx) :] else: self.close.append(self.open.pop(0)) return True, self.close[-1:] else: return False, -1
[docs] def optimal(self, groups): # see DIRECT Optimization Algorithm User Guide Daniel E. Finkel # for explanation # Potentially optimal index potoptidx = [] group_size = 0 for key, value in groups: subgroup = list(value) current_score = subgroup[0].score idx = 0 while ( idx < len(subgroup) and np.abs(subgroup[idx].score - current_score) <= 1e-13 ): current_score = subgroup[idx].score selected = subgroup[idx] current_idx = group_size + idx for jdx in range(current_idx + 1, len(self.open)): c = self.open[jdx] if c.length < selected.length: denom = selected.length - c.length num = selected.score - c.score if denom != 0: low_k = (num) / (denom) else: low_k = -float("inf") if low_k > self.maxi1[current_idx]: self.maxi1[current_idx] = low_k elif low_k < self.mini2[jdx]: self.mini2[jdx] = low_k elif c.length > selected.length: denom = c.length - selected.length num = c.score - selected.score if denom != 0: up_k = (num) / (denom) else: up_k = float("inf") if up_k < self.mini2[current_idx]: self.mini2[current_idx] = up_k elif up_k > self.maxi1[jdx]: self.maxi1[jdx] = up_k if self.mini2[current_idx] > 0 and ( self.maxi1[current_idx] <= self.mini2[current_idx] ): if self.best_score != 0: num = self.best_score - selected.score denum = np.abs(self.best_score) scnd_part = ( selected.length / denum * self.mini2[current_idx] ) if self.error <= num / denum + scnd_part: potoptidx.append(current_idx) else: scnd_part = selected.length * self.mini2[current_idx] if selected.score <= scnd_part: potoptidx.append(current_idx) idx += 1 group_size += len(subgroup) return potoptidx
[docs]class Locally_biased_POR(Tree_search): """Locally_biased_POR Potentially Optimal Rectangle algorithm (POR), is a the selection strategy comming from DIRECT. Attributes ---------- open : list[Fractal] Initial Open list containing not explored nodes from the partition tree. max_depth : int maximum depth of the partition tree. Q : int, default=1 Q-Best_first_search, at each get_next, tries to return Q nodes. reverse : boolean, default=False if False do a descending sort the open list, else do an ascending sort Methods ------- add(self,c) Add a node c to the fractal tree get_next(self) Get the next node to evaluate See Also -------- Fractal : Abstract class defining what a fractal is. FDA : Fractal Decomposition Algorithm Tree_search : Base class Beam_search : Memory efficient tree search algorithm based on BestFS Cyclic_best_first_search : Hybrid between DFS and BestFS """ def __init__(self, open, max_depth=600, error=1e-4, maxdiv=3000): """__init__(self, open, max_depth, Q=1, reverse=False, error=1e-4) Parameters ---------- open : list[Fractal] Initial Open list containing not explored nodes from the partition tree. max_depth : int maximum depth of the partition tree. Q : int, default=1 Q-Best_first_search, at each get_next, tries to return Q nodes. reverse : boolean, default=False if False do a descending sort the open list, else do an ascending sort error : float, default=1e-4 Small value which determines when an evaluation should be considered as good as the best solution found so far. """ super().__init__(open, max_depth) ############## # PARAMETERS # ############## self.error = error self.maxdiv = maxdiv ############# # VARIABLES # ############# self.maxi1 = np.full(self.maxdiv, -float("inf"), dtype=float) self.mini2 = np.full(self.maxdiv, float("inf"), dtype=float) self.next_frontier = [] min = [c.score for c in self.open] self.best_score = np.min(min)
[docs] def add(self, c): self.next_frontier.append(c) self.best_score = c.loss.best_score
[docs] def get_next(self): if len(self.next_frontier) > 0: self.open += sorted( self.next_frontier, key=lambda x: (-x.length, x.score) ) self.open = sorted(self.open, key=lambda x: (-x.length, x.score))[ : self.maxdiv ] self.next_frontier = [] if len(self.open): self.maxi1.fill(-float("inf")) self.mini2.fill(float("inf")) groups = groupby(self.open, lambda x: x.length) idx = self.optimal(groups) if idx: for i in reversed(idx): self.close.append(self.open.pop(i)) return True, self.close[-len(idx) :] else: self.close.append(self.open.pop(0)) return True, self.close[-1:] else: return False, -1
[docs] def optimal(self, groups): # Potentially optimal index potoptidx = defaultdict(None) group_size = 0 for key, value in groups: subgroup = list(value) current_score = subgroup[0].score idx = 0 while ( idx < len(subgroup) and np.abs(subgroup[idx].score - current_score) <= 1e-13 ): current_score = subgroup[idx].score selected = subgroup[idx] current_idx = group_size + idx for jdx in range(current_idx + 1, len(self.open)): c = self.open[jdx] if c.length < selected.length: denom = selected.length - c.length num = selected.score - c.score if denom != 0: low_k = (num) / (denom) else: low_k = -float("inf") if low_k > self.maxi1[current_idx]: self.maxi1[current_idx] = low_k elif low_k < self.mini2[jdx]: self.mini2[jdx] = low_k elif c.length > selected.length: denom = c.length - selected.length num = c.score - selected.score if denom != 0: up_k = (num) / (denom) else: up_k = float("inf") if up_k < self.mini2[current_idx]: self.mini2[current_idx] = up_k elif up_k > self.maxi1[jdx]: self.maxi1[jdx] = up_k if not (selected.length in potoptidx): if self.mini2[current_idx] > 0 and ( self.maxi1[current_idx] <= self.mini2[current_idx] ): if self.best_score != 0: num = self.best_score - selected.score denum = np.abs(self.best_score) scnd_part = ( selected.length / denum * self.mini2[current_idx] ) if self.error <= num / denum + scnd_part: potoptidx[selected.length] = current_idx else: scnd_part = ( selected.length * self.mini2[current_idx] ) if selected.score <= scnd_part: potoptidx[selected.length] = current_idx idx += 1 group_size += len(subgroup) return list(potoptidx.values())
[docs]class Adaptive_POR(Tree_search): """Adaptive_POR Adaptive_POR, is a the selection strategy comming from DIRECT-Restart. Attributes ---------- open : list[Fractal] Initial Open list containing not explored nodes from the partition tree. max_depth : int maximum depth of the partition tree. Q : int, default=1 Q-Best_first_search, at each get_next, tries to return Q nodes. reverse : boolean, default=False if False do a descending sort the open list, else do an ascending sort Methods ------- add(self,c) Add a node c to the fractal tree get_next(self) Get the next node to evaluate See Also -------- Fractal : Abstract class defining what a fractal is. FDA : Fractal Decomposition Algorithm Tree_search : Base class Beam_search : Memory efficient tree search algorithm based on BestFS Cyclic_best_first_search : Hybrid between DFS and BestFS """ def __init__( self, open, max_depth=600, error=1e-2, maxdiv=3000, patience=5 ): """__init__(self, open, max_depth, Q=1, reverse=False, error=1e-4) Parameters ---------- open : list[Fractal] Initial Open list containing not explored nodes from the partition tree. max_depth : int maximum depth of the partition tree. Q : int, default=1 Q-Best_first_search, at each get_next, tries to return Q nodes. reverse : boolean, default=False if False do a descending sort the open list, else do an ascending sort error : float, default=1e-4 Small value which determines when an evaluation should be considered as good as the best solution found so far. """ super().__init__(open, max_depth) ############## # PARAMETERS # ############## self.max_error = error self.maxdiv = maxdiv self.patience = patience ############# # VARIABLES # ############# self.maxi1 = np.full(self.maxdiv, -float("inf"), dtype=float) self.mini2 = np.full(self.maxdiv, float("inf"), dtype=float) self.next_frontier = [] min = [c.score for c in self.open] self.best_score = np.min(min) self.new_best_score = float("inf") self.stagnation = 0 self.error = self.max_error
[docs] def add(self, c): self.next_frontier.append(c) if c.loss.best_score < self.new_best_score: self.new_best_score = c.loss.best_score
[docs] def get_next(self): if len(self.next_frontier) > 0: self.open += sorted( self.next_frontier, key=lambda x: (-x.length, x.score) ) self.open = sorted(self.open, key=lambda x: (-x.length, x.score))[ : self.maxdiv ] self.next_frontier = [] if self.best_score - self.new_best_score >= 1e-4 * np.abs( np.median(self.open[0].loss.all_scores) - self.best_score ): self.best_score = self.new_best_score self.new_best_score = float("inf") self.stagnation = 0 else: self.stagnation += 1 if self.stagnation == self.patience: if self.error == 0.0: self.error = self.max_error else: self.error = 0.0 if len(self.open) > 1: self.maxi1.fill(-float("inf")) self.mini2.fill(float("inf")) groups = groupby(self.open, lambda x: x.length) idx = self.optimal(groups) if idx: for i in reversed(idx): self.close.append(self.open.pop(i)) return True, self.close[-len(idx) :] elif len(self.open) == 1: self.close.append(self.open.pop(0)) return True, self.close[-1:] else: return False, -1
[docs] def optimal(self, groups): # Potentially optimal index potoptidx = [] group_size = 0 for key, value in groups: subgroup = list(value) current_score = subgroup[0].score idx = 0 while ( idx < len(subgroup) and np.abs(subgroup[idx].score - current_score) <= 1e-13 ): current_score = subgroup[idx].score selected = subgroup[idx] current_idx = group_size + idx for jdx in range(current_idx + 1, len(self.open)): c = self.open[jdx] if c.length < selected.length: denom = selected.length - c.length num = selected.score - c.score if denom != 0: low_k = (num) / (denom) else: low_k = -float("inf") if low_k > self.maxi1[current_idx]: self.maxi1[current_idx] = low_k elif low_k < self.mini2[jdx]: self.mini2[jdx] = low_k elif c.length > selected.length: denom = c.length - selected.length num = c.score - selected.score if denom != 0: up_k = (num) / (denom) else: up_k = float("inf") if up_k < self.mini2[current_idx]: self.mini2[current_idx] = up_k elif up_k > self.maxi1[jdx]: self.maxi1[jdx] = up_k if self.mini2[current_idx] > 0 and ( self.maxi1[current_idx] <= self.mini2[current_idx] ): if self.best_score != 0: num = self.best_score - selected.score denum = np.abs(self.best_score) scnd_part = ( selected.length / denum * self.mini2[current_idx] ) if self.error <= num / denum + scnd_part: potoptidx.append(current_idx) else: scnd_part = selected.length * self.mini2[current_idx] if selected.score <= scnd_part: potoptidx.append(current_idx) idx += 1 group_size += len(subgroup) return potoptidx
class Potentially_Optimal_Hypersphere(Tree_search): """Potentially_Optimal_Hypersphere Potentially Optimal Hypersphere algorithm (POH), is a the selection strategy comming from DIRECT adapted for Hyperspheres. Attributes ---------- open : list[Fractal] Initial Open list containing not explored nodes from the partition tree. max_depth : int maximum depth of the partition tree. Q : int, default=1 Q-Best_first_search, at each get_next, tries to return Q nodes. reverse : boolean, default=False if False do a descending sort the open list, else do an ascending sort Methods ------- add(self,c) Add a node c to the fractal tree get_next(self) Get the next node to evaluate See Also -------- Fractal : Abstract class defining what a fractal is. FDA : Fractal Decomposition Algorithm Tree_search : Base class Beam_search : Memory efficient tree search algorithm based on BestFS Cyclic_best_first_search : Hybrid between DFS and BestFS """ def __init__(self, open, max_depth=600, error=1e-4, maxdiv=3000): """__init__(self, open, max_depth, Q=1, reverse=False, error=1e-4) Parameters ---------- open : list[Fractal] Initial Open list containing not explored nodes from the partition tree. max_depth : int maximum depth of the partition tree. Q : int, default=1 Q-Best_first_search, at each get_next, tries to return Q nodes. reverse : boolean, default=False if False do a descending sort the open list, else do an ascending sort error : float, default=1e-4 Small value which determines when an evaluation should be considered as good as the best solution found so far. """ super().__init__(open, max_depth) ############## # PARAMETERS # ############## self.error = error self.maxdiv = maxdiv ############# # VARIABLES # ############# self.maxi1 = np.full(self.maxdiv, -float("inf"), dtype=float) self.mini2 = np.full(self.maxdiv, float("inf"), dtype=float) self.next_frontier = [] min = [c.score for c in self.open] self.best_score = np.min(min) def add(self, c): self.next_frontier.append(c) self.best_score = c.loss.best_score def get_next(self): if len(self.next_frontier) > 0: # sort potentially optimal rectangle by radius (incresing) # then by score self.open += sorted( self.next_frontier, key=lambda x: (-x.radius, x.score) ) # clip open list to maxdiv self.open = sorted(self.open, key=lambda x: (-x.radius, x.score))[ : self.maxdiv ] self.next_frontier = [] if len(self.open): self.maxi1.fill(-float("inf")) self.mini2.fill(float("inf")) groups = groupby(self.open, lambda x: x.radius) idx = self.optimal(groups) if idx: for i in reversed(idx): self.close.append(self.open.pop(i)) return True, self.close[-len(idx) :] else: self.close.append(self.open.pop(0)) return True, self.close[-1:] else: return False, -1 def optimal(self, groups): # see DIRECT Optimization Algorithm User Guide Daniel E. Finkel # for explanation # Potentially optimal index potoptidx = [] group_size = 0 for key, value in groups: subgroup = list(value) current_score = subgroup[0].score idx = 0 while ( idx < len(subgroup) and np.abs(subgroup[idx].score - current_score) <= 1e-13 ): current_score = subgroup[idx].score selected = subgroup[idx] current_idx = group_size + idx for jdx in range(current_idx + 1, len(self.open)): c = self.open[jdx] if c.radius < selected.radius: denom = selected.radius - c.radius num = selected.score - c.score if denom != 0: low_k = (num) / (denom) else: low_k = -float("inf") if low_k > self.maxi1[current_idx]: self.maxi1[current_idx] = low_k elif low_k < self.mini2[jdx]: self.mini2[jdx] = low_k elif c.radius > selected.radius: denom = c.radius - selected.radius num = c.score - selected.score if denom != 0: up_k = (num) / (denom) else: up_k = float("inf") if up_k < self.mini2[current_idx]: self.mini2[current_idx] = up_k elif up_k > self.maxi1[jdx]: self.maxi1[jdx] = up_k if self.mini2[current_idx] > 0 and ( self.maxi1[current_idx] <= self.mini2[current_idx] ): if self.best_score != 0: num = self.best_score - selected.score denum = np.abs(self.best_score) scnd_part = ( selected.radius / denum * self.mini2[current_idx] ) if self.error <= num / denum + scnd_part: potoptidx.append(current_idx) else: scnd_part = selected.radius * self.mini2[current_idx] if selected.score <= scnd_part: potoptidx.append(current_idx) idx += 1 group_size += len(subgroup) return potoptidx ####### # SOO # ####### ####### # FDA # #######
[docs]class Move_up(Tree_search): """Move_up FDA tree search. Attributes ---------- open : list[Fractal] Initial Open list containing not explored nodes from the partition tree. max_depth : int maximum depth of the partition tree. Q : int, default=1 Q-Depth_first_search, at each get_next, tries to return Q nodes. reverse : boolean, default=False if False do a descending sort the open list, else do an ascending sort Methods ------- add(self,c) Add a node c to the fractal tree get_next(self) Get the next node to evaluate See Also -------- Fractal : Abstract class defining what a fractal is. FDA : Fractal Decomposition Algorithm Tree_search : Base class Breadth_first_search : Tree search Breadth based startegy Cyclic_best_first_search : Hybrid between DFS and BestFS """ def __init__(self, open, max_depth, Q=1, reverse=False): """__init__(open, max_depth, Q=1, reverse=False) Parameters ---------- open : list[Fractal] Initial Open list containing not explored nodes from the partition tree. max_depth : int maximum depth of the partition tree. Q : int, default=1 Q-Depth_first_search, at each get_next, tries to return Q nodes. reverse : boolean, default=False if False do a descending sort the open list, else do an ascending sort """ super().__init__(open, max_depth) ############## # PARAMETERS # ############## self.reverse = reverse self.Q = Q ############# # VARIABLES # ############# self.next_frontier = []
[docs] def add(self, c): self.next_frontier.append(c)
[docs] def get_next(self): if len(self.next_frontier) > 0: self.open = sorted( self.next_frontier + sorted( self.open, reverse=self.reverse, key=lambda x: (-x.level, x.score), ), reverse=self.reverse, key=lambda x: (-x.level, x.score), )[:] self.next_frontier = [] if len(self.open) > 0: for _ in range(self.Q): self.close.append(self.open.pop(0)) return True, self.close[-self.Q :] else: return False, -1