Testing stop_criteria

pygad/pygad.py

@@ -1171,23 +1171,40 @@ def __init__(self,
                 # reach_{target_fitness}: Stop if the target fitness value is reached.
                 # saturate_{num_generations}: Stop if the fitness value does not change (saturates) for the given number of generations.
                 criterion = stop_criteria.split("_")
-                if len(criterion) == 2:
-                    stop_word = criterion[0]
-                    number = criterion[1]
-
-                    if stop_word in self.supported_stop_words:
-                        pass
-                    else:
-                        self.valid_parameters = False
-                        raise ValueError(f"In the 'stop_criteria' parameter, the supported stop words are '{self.supported_stop_words}' but '{stop_word}' found.")
+                stop_word = criterion[0]
+                # criterion[1] might be a single or multiple numbers.
+                number = criterion[1:]
+                if stop_word in self.supported_stop_words:
+                    pass
+                else:
+                    self.valid_parameters = False
+                    raise ValueError(f"In the 'stop_criteria' parameter, the supported stop words are '{self.supported_stop_words}' but '{stop_word}' found.")
 
+                if len(criterion) == 2:
+                    # There is only a single number.
+                    number = number[0]
                     if number.replace(".", "").isnumeric():
                         number = float(number)
                     else:
                         self.valid_parameters = False
                         raise ValueError(f"The value following the stop word in the 'stop_criteria' parameter must be a number but the value ({number}) of type {type(number)} found.")
 
                     self.stop_criteria.append([stop_word, number])
+                elif len(criterion) > 2:
+                    if stop_word == 'reach':
+                        pass
+                    else:
+                        self.valid_parameters = False
+                        raise ValueError(f"Passing multiple numbers following the keyword in the 'stop_criteria' parameter is expected only with the 'reach' keyword but the keyword ({stop_word}) found.")
+
+                    for idx, num in enumerate(number):
+                        if num.replace(".", "").isnumeric():
+                            number[idx] = float(num)
+                        else:
+                            self.valid_parameters = False
+                            raise ValueError(f"The value(s) following the stop word in the 'stop_criteria' parameter must be numeric but the value ({num}) of type {type(num)} found.")
+
+                    self.stop_criteria.append([stop_word] + number)
 
                 else:
                     self.valid_parameters = False
@@ -2133,15 +2150,56 @@ def run(self):
                 if not self.stop_criteria is None:
                     for criterion in self.stop_criteria:
                         if criterion[0] == "reach":
-                            if max(self.last_generation_fitness) >= criterion[1]:
+                            # Single-objective problem.
+                            if type(self.last_generation_fitness[0]) in GA.supported_int_float_types:
+                                if max(self.last_generation_fitness) >= criterion[1]:
+                                    stop_run = True
+                                    break
+                            # Multi-objective problem.
+                            elif type(self.last_generation_fitness[0]) in [list, tuple, numpy.ndarray]:
+                                # Validate the value passed to the criterion.
+                                if len(criterion[1:]) == 1:
+                                    # There is a single value used across all the objectives.
+                                    pass
+                                elif len(criterion[1:]) > 1:
+                                    # There are multiple values. The number of values must be equal to the number of objectives.
+                                    if len(criterion[1:]) == len(self.last_generation_fitness[0]):
+                                        pass
+                                    else:
+                                        self.valid_parameters = False
+                                        raise ValueError(f"When the the 'reach' keyword is used with the 'stop_criteria' parameter for solving a multi-objective problem, then the number of numeric values following the keyword can be:\n1) A single numeric value to be used across all the objective functions.\n2) A number of numeric values equal to the number of objective functions.\nBut the value {criterion} found with {len(criterion)-1} numeric values which is not equal to the number of objective functions {len(self.last_generation_fitness[0])}.")
+
                                 stop_run = True
-                                break
+                                for obj_idx in range(len(self.last_generation_fitness[0])):
+                                    # Use the objective index to return the proper value for the criterion.
+
+                                    if len(criterion[1:]) == len(self.last_generation_fitness[0]):
+                                        reach_fitness_value = criterion[obj_idx + 1]
+                                    elif len(criterion[1:]) == 1:
+                                        reach_fitness_value = criterion[1]
+
+                                    if max(self.last_generation_fitness[:, obj_idx]) >= reach_fitness_value:
+                                        pass
+                                    else:
+                                        stop_run = False
+                                        break
                         elif criterion[0] == "saturate":
                             criterion[1] = int(criterion[1])
                             if (self.generations_completed >= criterion[1]):
-                                if (self.best_solutions_fitness[self.generations_completed - criterion[1]] - self.best_solutions_fitness[self.generations_completed - 1]) == 0:
+                                # Single-objective problem.
+                                if type(self.last_generation_fitness[0]) in GA.supported_int_float_types:
+                                    if (self.best_solutions_fitness[self.generations_completed - criterion[1]] - self.best_solutions_fitness[self.generations_completed - 1]) == 0:
+                                        stop_run = True
+                                        break
+                                # Multi-objective problem.
+                                elif type(self.last_generation_fitness[0]) in [list, tuple, numpy.ndarray]:
                                     stop_run = True
-                                    break
+                                    for obj_idx in range(len(self.last_generation_fitness[0])):
+                                        if (self.best_solutions_fitness[self.generations_completed - criterion[1]][obj_idx] - self.best_solutions_fitness[self.generations_completed - 1][obj_idx]) == 0:
+                                            pass
+                                        else:
+                                            stop_run = False
+                                            break
 
                 if stop_run:
                     break

tests/test_stop_criteria.py

@@ -0,0 +1,215 @@
+import pygad
+import numpy
+
+actual_num_fitness_calls_default_keep = 0
+actual_num_fitness_calls_no_keep = 0
+actual_num_fitness_calls_keep_elitism = 0
+actual_num_fitness_calls_keep_parents = 0
+
+num_generations = 100
+sol_per_pop = 10
+num_parents_mating = 5
+
+def multi_objective_problem(keep_elitism=1, 
+                            keep_parents=-1,
+                            fitness_batch_size=None,
+                            stop_criteria=None,
+                            parent_selection_type='sss',
+                            mutation_type="random",
+                            mutation_percent_genes="default",
+                            multi_objective=False):
+
+    function_inputs1 = [4,-2,3.5,5,-11,-4.7] # Function 1 inputs.
+    function_inputs2 = [-2,0.7,-9,1.4,3,5] # Function 2 inputs.
+    desired_output1 = 50 # Function 1 output.
+    desired_output2 = 30 # Function 2 output.
+
+    def fitness_func_batch_multi(ga_instance, solution, solution_idx):
+        f = []
+        for sol in solution:
+            output1 = numpy.sum(sol*function_inputs1)
+            output2 = numpy.sum(sol*function_inputs2)
+            fitness1 = 1.0 / (numpy.abs(output1 - desired_output1) + 0.000001)
+            fitness2 = 1.0 / (numpy.abs(output2 - desired_output2) + 0.000001)
+            f.append([fitness1, fitness2])
+        return f
+
+    def fitness_func_no_batch_multi(ga_instance, solution, solution_idx):
+        output1 = numpy.sum(solution*function_inputs1)
+        output2 = numpy.sum(solution*function_inputs2)
+        fitness1 = 1.0 / (numpy.abs(output1 - desired_output1) + 0.000001)
+        fitness2 = 1.0 / (numpy.abs(output2 - desired_output2) + 0.000001)
+        return [fitness1, fitness2]
+
+    def fitness_func_batch_single(ga_instance, solution, solution_idx):
+        f = []
+        for sol in solution:
+            output = numpy.sum(solution*function_inputs1)
+            fitness = 1.0 / (numpy.abs(output - desired_output1) + 0.000001)
+            f.append(fitness)
+        return f
+
+    def fitness_func_no_batch_single(ga_instance, solution, solution_idx):
+        output = numpy.sum(solution*function_inputs1)
+        fitness = 1.0 / (numpy.abs(output - desired_output1) + 0.000001)
+        return fitness
+
+    if fitness_batch_size is None or (type(fitness_batch_size) in pygad.GA.supported_int_types and fitness_batch_size == 1):
+        if multi_objective == True:
+            fitness_func = fitness_func_no_batch_multi
+        else:
+            fitness_func = fitness_func_no_batch_single
+    elif (type(fitness_batch_size) in pygad.GA.supported_int_types and fitness_batch_size > 1):
+        if multi_objective == True:
+            fitness_func = fitness_func_batch_multi
+        else:
+            fitness_func = fitness_func_batch_single
+
+    ga_optimizer = pygad.GA(num_generations=num_generations,
+                            sol_per_pop=sol_per_pop,
+                            num_genes=6,
+                            num_parents_mating=num_parents_mating,
+                            fitness_func=fitness_func,
+                            fitness_batch_size=fitness_batch_size,
+                            mutation_type=mutation_type,
+                            mutation_percent_genes=mutation_percent_genes,
+                            keep_elitism=keep_elitism,
+                            keep_parents=keep_parents,
+                            stop_criteria=stop_criteria,
+                            parent_selection_type=parent_selection_type,
+                            suppress_warnings=True)
+
+    ga_optimizer.run()
+
+    return ga_optimizer.generations_completed, ga_optimizer.best_solutions_fitness, ga_optimizer.last_generation_fitness, stop_criteria
+
+def test_number_calls_fitness_function_default_keep():
+    multi_objective_problem()
+
+def test_number_calls_fitness_function_stop_criteria_reach(multi_objective=False,
+                                                           fitness_batch_size=None, 
+                                                           num=10):
+    generations_completed, best_solutions_fitness, last_generation_fitness, stop_criteria = multi_objective_problem(multi_objective=multi_objective,
+                                                                                           fitness_batch_size=fitness_batch_size,
+                                                                                           stop_criteria=f"reach_{num}")
+    # Verify that the GA stops when meeting the criterion.
+    criterion = stop_criteria.split('_')
+    stop_word = criterion[0]
+    if generations_completed < num_generations:
+        if stop_word == 'reach':
+            if len(criterion) > 2:
+                # multi-objective problem.
+                for idx, num in enumerate(criterion[1:]):
+                    criterion[idx + 1] = float(num)
+            else:
+                criterion[1] = float(criterion[1])
+
+            # Single-objective
+            if type(last_generation_fitness[0]) in pygad.GA.supported_int_float_types:
+                assert max(last_generation_fitness) >= criterion[1]
+            # Multi-objective
+            elif type(last_generation_fitness[0]) in [list, tuple, numpy.ndarray]:
+                # Validate the value passed to the criterion.
+                if len(criterion[1:]) == 1:
+                    # There is a single value used across all the objectives.
+                    pass
+                elif len(criterion[1:]) > 1:
+                    # There are multiple values. The number of values must be equal to the number of objectives.
+                    if len(criterion[1:]) == len(last_generation_fitness[0]):
+                        pass
+                    else:
+                        raise ValueError("Error")
+
+                for obj_idx in range(len(last_generation_fitness[0])):
+                    # Use the objective index to return the proper value for the criterion.
+                    if len(criterion[1:]) == len(last_generation_fitness[0]):
+                        reach_fitness_value = criterion[obj_idx + 1]
+                    elif len(criterion[1:]) == 1:
+                        reach_fitness_value = criterion[1]
+
+                    assert max(last_generation_fitness[:, obj_idx]) >= reach_fitness_value
+
+def test_number_calls_fitness_function_stop_criteria_saturate(multi_objective=False,
+                                                              fitness_batch_size=None,
+                                                              num=5):
+    generations_completed, best_solutions_fitness, last_generation_fitness, stop_criteria = multi_objective_problem(multi_objective=multi_objective,
+                                                                                           fitness_batch_size=fitness_batch_size,
+                                                                                           stop_criteria=f"saturate_{num}")
+    # Verify that the GA stops when meeting the criterion.
+    criterion = stop_criteria.split('_')
+    stop_word = criterion[0]
+    number = criterion[1]
+    if generations_completed < num_generations:
+        if stop_word == 'saturate':
+            number = int(number)
+            if type(last_generation_fitness[0]) in pygad.GA.supported_int_float_types:
+                assert best_solutions_fitness[generations_completed - number] == best_solutions_fitness[generations_completed - 1]
+            elif type(last_generation_fitness[0]) in [list, tuple, numpy.ndarray]:
+                for obj_idx in range(len(best_solutions_fitness[0])):
+                    assert best_solutions_fitness[generations_completed - number][obj_idx] == best_solutions_fitness[generations_completed - 1][obj_idx]
+
+if __name__ == "__main__":
+    #### Single-objective problem with a single numeric value with stop_criteria.
+    print()
+    test_number_calls_fitness_function_default_keep()
+    print()
+    test_number_calls_fitness_function_stop_criteria_reach()
+    print()
+    test_number_calls_fitness_function_stop_criteria_reach(num=2)
+    print()
+    test_number_calls_fitness_function_stop_criteria_saturate()
+    print()
+    test_number_calls_fitness_function_stop_criteria_saturate(num=2)
+    print()
+    test_number_calls_fitness_function_stop_criteria_reach(fitness_batch_size=4)
+    print()
+    test_number_calls_fitness_function_stop_criteria_reach(fitness_batch_size=4,
+                                                           num=2)
+    print()
+    test_number_calls_fitness_function_stop_criteria_saturate(fitness_batch_size=4)
+    print()
+    test_number_calls_fitness_function_stop_criteria_saturate(fitness_batch_size=4,
+                                                              num=2)
+    print()
+
+
+    #### Multi-objective problem with a single numeric value with stop_criteria.
+    test_number_calls_fitness_function_stop_criteria_reach(multi_objective=True)
+    print()
+    test_number_calls_fitness_function_stop_criteria_reach(multi_objective=True, 
+                                                           num=2)
+    print()
+    test_number_calls_fitness_function_stop_criteria_saturate(multi_objective=True)
+    print()
+    test_number_calls_fitness_function_stop_criteria_saturate(multi_objective=True, 
+                                                              num=2)
+    print()
+    test_number_calls_fitness_function_stop_criteria_reach(multi_objective=True, 
+                                                           fitness_batch_size=4)
+    print()
+    test_number_calls_fitness_function_stop_criteria_reach(multi_objective=True, 
+                                                           fitness_batch_size=4,
+                                                           num=2)
+    print()
+    test_number_calls_fitness_function_stop_criteria_saturate(multi_objective=True, 
+                                                              fitness_batch_size=4)
+    print()
+    test_number_calls_fitness_function_stop_criteria_saturate(multi_objective=True, 
+                                                              fitness_batch_size=4,
+                                                              num=50)
+    print()
+
+
+    #### Multi-objective problem with multiple numeric values with stop_criteria.
+    test_number_calls_fitness_function_stop_criteria_reach(multi_objective=True)
+    print()
+    test_number_calls_fitness_function_stop_criteria_reach(multi_objective=True, 
+                                                           num="2_5")
+    print()
+    test_number_calls_fitness_function_stop_criteria_reach(multi_objective=True, 
+                                                           fitness_batch_size=4)
+    print()
+    test_number_calls_fitness_function_stop_criteria_reach(multi_objective=True, 
+                                                           fitness_batch_size=4,
+                                                           num="10_20")
+