Merge pull request #65 from alcides/metahandlers

Metahandlers

aeon/__main__.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 import sys
 
 import argparse
@@ -17,6 +18,9 @@
 from aeon.sugar.desugar import desugar
 from aeon.sugar.parser import parse_program
 from aeon.sugar.program import Program
+from aeon.synthesis.uis.api import SynthesisUI
+from aeon.synthesis.uis.ncurses import NCursesUI
+from aeon.synthesis.uis.terminal import TerminalUI
 from aeon.synthesis_grammar.identification import incomplete_functions_and_holes
 from aeon.synthesis_grammar.synthesizer import synthesize, parse_config
 from aeon.typechecking.typeinfer import check_type_errors
@@ -75,6 +79,13 @@ def parse_arguments():
         action="store_true",
         help="Show timing information",
     )
+
+    parser.add_argument(
+        "-rg",
+        "--refined-grammar",
+        action="store_true",
+        help="Use the refined grammar for synthesis",
+    )
     return parser.parse_args()
 
 
@@ -92,6 +103,13 @@ def log_type_errors(errors: list[Exception | str]):
     print("TYPECHECKER", "-------------------------------")
 
 
+def select_synthesis_ui() -> SynthesisUI:
+    if os.environ.get("TERM", None):
+        return NCursesUI()
+    else:
+        return TerminalUI()
+
+
 def main() -> None:
     args = parse_arguments()
     logger = setup_logger()
@@ -145,19 +163,21 @@ def main() -> None:
                             if args.gp_config and args.config_section else
                             None)
 
+        ui = select_synthesis_ui()
+
         with RecordTime("Synthesis"):
-            synthesis_result = synthesize(
+            program, terms = synthesize(
                 typing_ctx,
                 evaluation_ctx,
                 core_ast_anf,
                 incomplete_functions,
                 metadata,
                 filename,
                 synth_config,
+                args.refined_grammar,
+                ui,
             )
-            print(f"{synthesis_result}")
-            # print()
-            # pretty_print_term(ensure_anf(synthesis_result, 200))
+            ui.display_results(program, terms)
         sys.exit(0)
     with RecordTime("Evaluation"):
         eval(core_ast, evaluation_ctx)

aeon/core/substitutions.py

@@ -89,29 +89,25 @@ def rec(x: Term):
     assert False
 
 
-def substitution_in_liquid(t: LiquidTerm, rep: LiquidTerm,
-                           name: str) -> LiquidTerm:
+def substitution_in_liquid(t: LiquidTerm, rep: LiquidTerm, name: str) -> LiquidTerm:
     """substitutes name in the term t with the new replacement term rep."""
     assert isinstance(rep, LiquidTerm)
-    if isinstance(t, (LiquidLiteralInt, LiquidLiteralBool, LiquidLiteralString,
-                      LiquidLiteralFloat)):
+    if isinstance(t, (LiquidLiteralInt, LiquidLiteralBool, LiquidLiteralString, LiquidLiteralFloat)):
         return t
     elif isinstance(t, LiquidVar):
         if t.name == name:
             return rep
         else:
             return t
     elif isinstance(t, LiquidApp):
-        return LiquidApp(
-            t.fun, [substitution_in_liquid(a, rep, name) for a in t.args])
+        return LiquidApp(t.fun, [substitution_in_liquid(a, rep, name) for a in t.args])
     elif isinstance(t, LiquidHole):
         if t.name == name:
             return rep
         else:
             return LiquidHole(
                 t.name,
-                [(substitution_in_liquid(a, rep, name), t)
-                 for (a, t) in t.argtypes],
+                [(substitution_in_liquid(a, rep, name), t) for (a, t) in t.argtypes],
             )
     else:
         print(t, type(t))
@@ -223,16 +219,15 @@ def liquefy_app(app: Application) -> LiquidApp | None:
     elif isinstance(app.fun, Application):
         liquid_pseudo_fun = liquefy_app(app.fun)
         if liquid_pseudo_fun:
-            return LiquidApp(liquid_pseudo_fun.fun,
-                             liquid_pseudo_fun.args + [arg])
+            return LiquidApp(liquid_pseudo_fun.fun, liquid_pseudo_fun.args + [arg])
         return None
     elif isinstance(app.fun, Let):
         return liquefy_app(
             Application(
-                substitution(app.fun.body, app.fun.var_value,
-                             app.fun.var_name),
+                substitution(app.fun.body, app.fun.var_value, app.fun.var_name),
                 app.arg,
-            ), )
+            ),
+        )
     assert False
 
 

aeon/core/types.py

@@ -223,4 +223,6 @@ def args_size_of_type(t: Type) -> int:
 def refined_to_unrefined_type(ty: Type) -> Type:
     if isinstance(ty, RefinedType):
         return ty.type
+    if isinstance(ty, AbstractionType):
+        return AbstractionType(ty.var_name, refined_to_unrefined_type(ty.var_type), refined_to_unrefined_type(ty.type))
     return ty

aeon/synthesis/uis/api.py

@@ -0,0 +1,58 @@
+import abc
+from typing import Any
+
+from aeon.backend.evaluator import EvaluationContext
+from aeon.core.terms import Term
+from aeon.core.types import Type
+from aeon.typechecking.context import TypingContext
+
+
+class SynthesisUI(abc.ABC):
+
+    def start(
+        self,
+        typing_ctx: TypingContext,
+        evaluation_ctx: EvaluationContext,
+        target_name: str,
+        target_type: Type,
+        budget: Any,
+    ):
+        ...
+
+    def register(self, solution: Term, quality: Any, elapsed_time: float,
+                 is_best: bool):
+        ...
+
+    def end(self, solution: Term, quality: Any):
+        ...
+
+    def wrapper(self, f):
+        """This wrapper is necessary for the NCurses version of the API"""
+        return f()
+
+    def display_results(self, program: Term, terms: dict[str, Term]):
+        print("Synthesized holes:")
+        for name in terms:
+            print(f"?{name}: {terms[name]}")
+        # print()
+        # pretty_print_term(ensure_anf(synthesis_result, 200))
+
+
+class SilentSynthesisUI(SynthesisUI):
+
+    def start(
+        self,
+        typing_ctx: TypingContext,
+        evaluation_ctx: EvaluationContext,
+        target_name: str,
+        target_type: Type,
+        budget: Any,
+    ):
+        pass
+
+    def register(self, solution: Term, quality: Any, elapsed_time: float,
+                 is_best: bool):
+        pass
+
+    def end(self, solution: Term, quality: Any):
+        pass

aeon/synthesis/uis/ncurses.py

@@ -0,0 +1,49 @@
+from typing import Any
+from aeon.backend.evaluator import EvaluationContext
+from aeon.core.terms import Term
+from aeon.core.types import Type
+from aeon.synthesis.uis.api import SynthesisUI
+from aeon.typechecking.context import TypingContext
+
+import curses
+
+
+class NCursesUI(SynthesisUI):
+
+    def start(
+        self,
+        typing_ctx: TypingContext,
+        evaluation_ctx: EvaluationContext,
+        target_name: str,
+        target_type: Type,
+        budget: Any,
+    ):
+        self.stdscr = curses.initscr()
+        self.target_name = target_name
+        self.target_type = target_type
+        self.budget = budget
+
+    def register(self, solution: Term, quality: Any, elapsed_time: float,
+                 is_best: bool):
+        if is_best:
+            self.best_solution = solution
+            self.best_quality = quality
+
+        self.stdscr.clear()
+        self.stdscr.addstr(0, 0, f"Synthesizing ?{self.target_name}")
+        self.stdscr.addstr(1, 0, "====================================")
+        self.stdscr.addstr(3, 0, f"Fitness: {quality}")
+        self.stdscr.addstr(4, 0, f"Program: {solution}")
+        self.stdscr.addstr(6, 0, f"Best: {self.best_solution}")
+        self.stdscr.addstr(7, 0, f"Best Fitness: {self.best_quality}")
+        self.stdscr.addstr(9, 0, "====================================")
+        self.stdscr.addstr(10, 0,
+                           f"""{elapsed_time:.1f} / {self.budget:.1f}s""")
+
+        self.stdscr.refresh()
+
+    def end(self, solution: Term, quality: Any):
+        curses.endwin()
+
+    def wrapper(self, f):
+        return curses.wrapper(f)

aeon/synthesis/uis/terminal.py

@@ -0,0 +1,35 @@
+from typing import Any
+from aeon.backend.evaluator import EvaluationContext
+from aeon.core.terms import Term
+from aeon.core.types import Type
+from aeon.synthesis.uis.api import SynthesisUI
+from aeon.typechecking.context import TypingContext
+
+
+class TerminalUI(SynthesisUI):
+
+    def start(
+        self,
+        typing_ctx: TypingContext,
+        evaluation_ctx: EvaluationContext,
+        target_name: str,
+        target_type: Type,
+        budget: Any,
+    ):
+        self.target_name = target_name
+        self.target_type = target_type
+        self.budget = budget
+
+    def register(self, solution: Term, quality: Any, elapsed_time: float,
+                 is_best: bool):
+        if is_best:
+            self.best_solution = solution
+            self.best_quality = quality
+
+        print(
+            f"Target: {self.target_name} ({elapsed_time:.1f} / {self.budget:.1f}s) "
+            + f"| Best: {self.best_solution} ({self.best_quality}) " +
+            f"| Current: {solution} ({quality})")
+
+    def end(self, solution: Term, quality: Any):
+        pass