Merge pull request #251 from alcides/enumerative

Enumerative algorithm for synthesis

docs/source/metahandlers/api.md

@@ -18,20 +18,12 @@ Here are a few MetaHandlers provided by GeneticEngine
 .. autoapiclass:: geneticengine.grammar.metahandlers.floats.FloatRange
 ```
 
-```{eval-rst}
-.. autoapiclass:: geneticengine.grammar.metahandlers.floats.FloatList
-```
-
 #### Lists
 
 ```{eval-rst}
 .. autoapiclass:: geneticengine.grammar.metahandlers.lists.ListSizeBetween
 ```
 
-```{eval-rst}
-.. autoapiclass:: geneticengine.grammar.metahandlers.lists.ListSizeBetweenWithoutListOperations
-```
-
 #### Strings
 
 ```{eval-rst}

examples/classification_unknown_length_objectives.py

@@ -19,7 +19,6 @@
 from geml.grammars.sgp import Number
 from geml.grammars.sgp import Plus
 from geml.grammars.sgp import Var
-from geneticengine.grammar.metahandlers.floats import FloatList
 from geneticengine.grammar.metahandlers.vars import VarRange
 from geneticengine.problems import LazyMultiObjectiveProblem
 from geneticengine.random.sources import NativeRandomSource
@@ -143,7 +142,7 @@ def __str__(self) -> str:
 
 @dataclass
 class Literal2(Number):
-    val: Annotated[float, FloatList([-1, -0.1, -0.01, -0.001, 1, 0.1, 0.01, 0.001])]
+    val: Annotated[float, VarRange([-1, -0.1, -0.01, -0.001, 1, 0.1, 0.01, 0.001])]
 
     def evaluate(self, **kwargs):
         return self.val

geneticengine/algorithms/api.py

@@ -7,33 +7,26 @@
     ProgressTracker,
 )
 from geneticengine.evaluation.sequential import SequentialEvaluator
-from geneticengine.problems import Problem, SingleObjectiveProblem
-from geneticengine.representations.api import Representation
+from geneticengine.problems import Problem
 from geneticengine.solutions.individual import Individual
 
 
 class SynthesisAlgorithm(ABC):
     tracker: ProgressTracker
     problem: Problem
     budget: SearchBudget
-    representation: Representation
 
     def __init__(
         self,
         problem: Problem,
         budget: SearchBudget,
-        representation: Representation,
         tracker: Optional[ProgressTracker] = None,
     ):
         self.problem = problem
         self.budget = budget
-        self.representation = representation
 
         if tracker is None:
-            if isinstance(problem, SingleObjectiveProblem):
-                self.tracker = ProgressTracker(problem, SequentialEvaluator())
-            else:
-                self.tracker = ProgressTracker(problem, SequentialEvaluator())
+            self.tracker = ProgressTracker(problem, SequentialEvaluator())
         else:
             self.tracker = tracker
 

geneticengine/algorithms/enumerative.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+from itertools import count, takewhile
+from typing import Any
+
+from geneticengine.algorithms.api import SynthesisAlgorithm
+
+from geneticengine.evaluation.budget import SearchBudget
+from geneticengine.evaluation.tracker import ProgressTracker
+from geneticengine.exceptions import GeneticEngineError
+from geneticengine.grammar.grammar import Grammar
+from geneticengine.grammar.metahandlers.base import MetaHandlerGenerator
+from geneticengine.grammar.utils import (
+    get_arguments,
+    get_generic_parameter,
+    get_generic_parameters,
+    is_generic_list,
+    is_generic_tuple,
+    is_metahandler,
+    is_union,
+)
+from geneticengine.problems import Problem
+from geneticengine.representations.tree.initializations import apply_constructor
+from geneticengine.solutions.individual import (
+    ConcreteIndividual,
+    Individual,
+)
+
+
+def frange(start, stop, step):
+    return takewhile(lambda x: x < stop, count(start, step))
+
+
+def combine_list_types(ts: list[type], acc: list[Any], gen):
+    match ts:
+        case []:
+            yield acc
+        case _:
+            head = ts[0]
+            tail = ts[1:]
+            for x in gen(head):
+                yield from combine_list_types(tail, acc + [x], gen)
+
+
+def iterate_grammar(grammar: Grammar, starting_symbol: type):
+    def rec_generator(symbol):
+        return iterate_grammar(grammar, symbol)
+
+    if starting_symbol is int:
+        yield from range(-100000000, 100000000)
+    elif starting_symbol is float:
+        yield from frange(-100000.0, 100000.0, 0.00001)
+    elif starting_symbol is bool:
+        yield True
+        yield False
+    elif is_generic_tuple(starting_symbol):
+        types = get_generic_parameters(starting_symbol)
+        for li in combine_list_types(types, [], rec_generator):
+            yield tuple(li)
+    elif is_generic_list(starting_symbol):
+        inner_type = get_generic_parameter(starting_symbol)
+
+        for length in range(0, 1024):
+
+            generator_list = [inner_type for _ in range(length)]
+            for concrete_list in combine_list_types(generator_list, [], rec_generator):
+                yield concrete_list
+
+    elif is_metahandler(starting_symbol):
+        metahandler: MetaHandlerGenerator = starting_symbol.__metadata__[0]  # type: ignore
+        base_type = get_generic_parameter(starting_symbol)
+
+        if hasattr(metahandler, "iterate"):
+            yield from metahandler.iterate(base_type, lambda xs: combine_list_types(xs, [], rec_generator))
+        else:
+            base_type = get_generic_parameter(starting_symbol)
+            for ins in iterate_grammar(grammar, base_type):
+                if metahandler.validate(ins):
+                    yield ins
+    elif is_union(starting_symbol):
+        for alt in get_generic_parameters(starting_symbol):
+            yield from iterate_grammar(grammar, alt)
+    else:
+        if starting_symbol not in grammar.all_nodes:
+            raise GeneticEngineError(
+                f"Symbol {starting_symbol} not in grammar rules.",
+            )
+        elif starting_symbol in grammar.alternatives:
+            compatible_productions = grammar.alternatives[starting_symbol]
+            for prod in compatible_productions:
+                yield from iterate_grammar(grammar, prod)
+        else:
+            # Normal production
+            args = []
+            # TODO: Add dependent types to enumerative
+            # dependent_values = {}
+            args = [argt for _, argt in get_arguments(starting_symbol)]
+            for li in combine_list_types(args, [], rec_generator):
+                yield apply_constructor(starting_symbol, li)
+
+
+def iterate_individuals(grammar: Grammar, starting_symbol: type):
+    for p in iterate_grammar(grammar, starting_symbol):
+        yield ConcreteIndividual(instance=p)
+
+
+class EnumerativeSearch(SynthesisAlgorithm):
+    """Iterates through all possible representations and selects the best."""
+
+    def __init__(
+        self,
+        problem: Problem,
+        budget: SearchBudget,
+        grammar: Grammar,
+        tracker: ProgressTracker | None = None,
+    ):
+        super().__init__(problem, budget, tracker)
+        self.grammar = grammar
+
+    def perform_search(self) -> list[Individual]:
+        for individual in iterate_individuals(self.grammar, self.grammar.starting_symbol):
+            self.tracker.evaluate_single(individual)
+            if self.is_done():
+                break
+        return self.tracker.get_best_individuals()

geneticengine/algorithms/gp/adaptive.py

@@ -14,7 +14,7 @@
 from geneticengine.problems import Fitness, Problem
 from geneticengine.random.sources import RandomSource
 from geneticengine.representations.api import Representation
-from geneticengine.solutions.individual import Individual
+from geneticengine.solutions.individual import PhenotypicIndividual
 
 
 def generate_random_population_size(random: RandomSource) -> int:
@@ -47,9 +47,9 @@ def initialize(
         random: RandomSource,
         target_size: int,
         **kwargs,
-    ) -> Iterator[Individual]:
+    ) -> Iterator[PhenotypicIndividual]:
         while not self.budget.is_done(self.tracker):
-            yield Individual(
+            yield PhenotypicIndividual(
                 representation.create_genotype(
                     random,
                     **kwargs,
@@ -73,7 +73,7 @@ def post_iterate(
         evaluator: Evaluator,
         representation: Representation,
         random: RandomSource,
-        population: Iterator[Individual],
+        population: Iterator[PhenotypicIndividual],
         target_size: int,
         generation: int,
     ) -> None:
@@ -108,11 +108,11 @@ def iterate(
         evaluator: Evaluator,
         representation: Representation,
         random: RandomSource,
-        population: Iterator[Individual],
+        population: Iterator[PhenotypicIndividual],
         target_size: int,
         generation: int,
-    ) -> Iterator[Individual]:
-        npopulation: list[Individual] = [i for i in population]
+    ) -> Iterator[PhenotypicIndividual]:
+        npopulation: list[PhenotypicIndividual] = [i for i in population]
         ranges = self.compute_ranges(npopulation, target_size)
         assert len(ranges) == len(self.steps)
 
@@ -141,7 +141,7 @@ def iterate(
                 yield from new_individuals
 
 
-def best_of_population(population: list[Individual], problem: Problem) -> Individual:
+def best_of_population(population: list[PhenotypicIndividual], problem: Problem) -> PhenotypicIndividual:
     return reduce(
         lambda x, s: x if problem.is_better(x.get_fitness(problem), s.get_fitness(problem)) else s,
         population,
@@ -162,7 +162,7 @@ def post_iterate(
         evaluator: Evaluator,
         representation: Representation,
         random: RandomSource,
-        population: Iterator[Individual],
+        population: Iterator[PhenotypicIndividual],
         target_size: int,
         generation: int,
     ) -> None:
@@ -193,7 +193,7 @@ def post_iterate(
         evaluator: Evaluator,
         representation: Representation,
         random: RandomSource,
-        population: Iterator[Individual],
+        population: Iterator[PhenotypicIndividual],
         target_size: int,
         generation: int,
     ) -> None:

geneticengine/algorithms/gp/operators/combinators.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 from typing import Iterator
 
-from geneticengine.solutions.individual import Individual
+from geneticengine.solutions.individual import Individual, PhenotypicIndividual
 from geneticengine.algorithms.gp.structure import GeneticStep
 from geneticengine.problems import Problem
 from geneticengine.random.sources import RandomSource
@@ -18,10 +18,10 @@ def iterate(
         evaluator: Evaluator,
         representation: Representation,
         random: RandomSource,
-        population: Iterator[Individual],
+        population: Iterator[PhenotypicIndividual],
         target_size: int,
         generation: int,
-    ) -> Iterator[Individual]:
+    ) -> Iterator[PhenotypicIndividual]:
         for _, p in zip(range(target_size), population):
             yield p
 
@@ -39,10 +39,10 @@ def iterate(
         evaluator: Evaluator,
         representation: Representation,
         random: RandomSource,
-        population: Iterator[Individual],
+        population: Iterator[PhenotypicIndividual],
         target_size: int,
         generation: int,
-    ) -> Iterator[Individual]:
+    ) -> Iterator[PhenotypicIndividual]:
         npopulation = population
         for step in self.steps:
             npopulation = step.apply(
@@ -101,10 +101,10 @@ def iterate(
         evaluator: Evaluator,
         representation: Representation,
         random: RandomSource,
-        population: Iterator[Individual],
+        population: Iterator[PhenotypicIndividual],
         target_size: int,
         generation: int,
-    ) -> Iterator[Individual]:
+    ) -> Iterator[PhenotypicIndividual]:
         npopulation: list[Individual] = [i for i in population]
         ranges = self.compute_ranges(npopulation, target_size)
         assert len(ranges) == len(self.steps)
@@ -149,11 +149,11 @@ def iterate(
         evaluator: Evaluator,
         representation: Representation,
         random: RandomSource,
-        population: Iterator[Individual],
+        population: Iterator[PhenotypicIndividual],
         target_size: int,
         generation: int,
-    ) -> Iterator[Individual]:
-        npopulation: list[Individual] = list(population)
+    ) -> Iterator[PhenotypicIndividual]:
+        npopulation: list[PhenotypicIndividual] = list(population)
         total = sum(self.weights)
         indices = [0] + self.cumsum(
             [int(round(w * len(npopulation) / total, 0)) for w in self.weights],

geneticengine/algorithms/gp/operators/crossover.py

@@ -2,7 +2,7 @@
 import logging
 from typing import Iterator
 
-from geneticengine.solutions.individual import Individual
+from geneticengine.solutions.individual import Individual, PhenotypicIndividual
 from geneticengine.algorithms.gp.structure import GeneticStep
 from geneticengine.problems import Problem
 from geneticengine.random.sources import RandomSource
@@ -28,10 +28,10 @@ def iterate(
         evaluator: Evaluator,
         representation: Representation,
         random: RandomSource,
-        population: Iterator[Individual],
+        population: Iterator[PhenotypicIndividual],
         target_size: int,
         generation: int,
-    ) -> Iterator[Individual]:
+    ) -> Iterator[PhenotypicIndividual]:
         assert isinstance(representation, RepresentationWithCrossover)
         npopulation = list(population)
         for i in range(target_size // 2):
@@ -53,8 +53,8 @@ def iterate(
     def crossover(
         self,
         random: RandomSource,
-        individual1: Individual,
-        individual2: Individual,
+        individual1: PhenotypicIndividual,
+        individual2: PhenotypicIndividual,
         representation: Representation,
     ):
         assert isinstance(representation, RepresentationWithCrossover)
@@ -65,6 +65,6 @@ def crossover(
             individual2.genotype,
         )
         return (
-            Individual(g1, representation=representation),
-            Individual(g2, representation=representation),
+            PhenotypicIndividual(g1, representation=representation),
+            PhenotypicIndividual(g2, representation=representation),
         )