Refactoring of Generators

README.md

@@ -159,11 +159,11 @@ You can download and compile these solvers from the [MaxSAT 2023 evaluation](htt
 You can import Popper and use it in your Python code like so:
 
 ```python
-from popper.util import Settings, print_prog_score
+from popper.util import Settings
 from popper.loop import learn_solution
 
 settings = Settings(kbpath='input_dir')
 prog, score, stats = learn_solution(settings)
 if prog != None:
-    print_prog_score(prog, score)
+    settings.print_prog_score(prog, score)
 ```

popper.py

@@ -6,7 +6,7 @@
 if __name__ == '__main__':
     settings = Settings(cmd_line=True)
     prog, score, stats = learn_solution(settings)
-    if prog != None:
+    if prog is not None:
         settings.print_prog_score(prog, score)
     else:
         print('NO SOLUTION')

popper/abs_generate.py

@@ -0,0 +1,56 @@
+import abc
+from typing import FrozenSet, Tuple, TYPE_CHECKING, List
+
+import clingo
+
+if TYPE_CHECKING:
+    from . util import Settings
+
+Rule = Tuple['Literal', FrozenSet['Literal']]
+RuleBase = FrozenSet[Rule]
+
+class Generator(abc.ABC):
+    settings: 'Settings'
+
+    # @profile
+    def get_prog(self) -> RuleBase:
+        pass
+
+    def gen_symbol(self, literal, backend):
+        pass
+
+    def update_solver(self, size):
+        pass
+        self.update_number_of_literals(size)
+
+    def update_number_of_literals(self, size):
+        pass
+
+    def update_number_of_vars(self, size):
+        pass
+
+    def update_number_of_rules(self, size):
+        pass
+
+    def prune_size(self, size):
+        pass
+
+    # @profile
+    def get_ground_rules(self, rule):
+        pass
+
+    def parse_handles(self, new_handles):
+        pass
+
+    def constrain(self, tmp_new_cons):
+        pass
+
+    @abc.abstractmethod
+    def build_encoding(self, bkcons: List, settings: "Settings") -> str:
+        """Build and return a string for an ASP solver, used to generate hypotheses."""
+        pass
+
+    @abc.abstractmethod
+    def init_solver(self, encoding: str) -> clingo.Control:
+        """Incorporate the `encoding` into a new solver (`clingo.Control`) and return it."""
+        pass

popper/bkcons.py

@@ -1,16 +1,16 @@
+import numbers
+import sys
+import traceback
+from collections import defaultdict
+from itertools import product
+from traceback import format_tb
+
 import clingo
 import clingo.script
-import numbers
-import operator
-import pkg_resources
-import time
-from . util import rule_is_recursive, format_rule, Constraint, order_prog, Literal, suppress_stdout_stderr
 from clingo import Function, Number, Tuple_
-from collections import defaultdict
-from itertools import permutations, product
-from pysat.card import *
-from pysat.formula import CNF
-from pysat.solvers import Solver
+
+from .util import suppress_stdout_stderr
+
 clingo.script.enable_python()
 
 tmp_map = {}
@@ -52,7 +52,8 @@
 #     return (head_pred, arity), body_preds
 
 
-from itertools import permutations, combinations
+from itertools import permutations
+
 all_myvars = ['A','B','C','D','E','F','G','H']
 
 def connected(xs, ys):
@@ -628,7 +629,6 @@ def atom_to_symbol(pred, args):
     return Function(name = pred, arguments = xs)
 
 def deduce_bk_cons(settings, tester):
-    import re
     prog = []
     lookup2 = {k: f'({v})' for k,v in tmp_map.items()}
     lookup1 = {k:v for k,v in lookup2.items()}
@@ -1110,4 +1110,4 @@ def deduce_non_singletons(settings):
         cons.append(con)
 
     # exit()
-    return cons
+    return cons

popper/combine.py

@@ -258,7 +258,7 @@ def find_combination(self, timeout):
 
             else:
                 if not self.settings.lex:
-                    print("ERROR: Combining rec or pi programs not supported with MDL objective. Exiting.")
+                    self.settings.logger.error("ERROR: Combining rec or pi programs not supported with MDL objective. Exiting.")
                     assert(False)
 
                 model_inconsistent = self.tester.test_prog_inconsistent(model_prog)

popper/gen2.py

@@ -1,12 +1,16 @@
-import time
+import numbers
 import re
+from itertools import permutations
+from typing import Any, Set, Tuple, Optional
+
 import clingo
-import numbers
 import clingo.script
-import pkg_resources
-from . util import Constraint, Literal
 from clingo import Function, Number, Tuple_
-from itertools import permutations
+
+from .util import Constraint, Literal, Settings
+from . abs_generate import Generator as AbstractGenerator
+from .resources import resource_string
+
 
 def arg_to_symbol(arg):
     if isinstance(arg, tuple):
@@ -15,10 +19,13 @@ def arg_to_symbol(arg):
         return Number(arg)
     if isinstance(arg, str):
         return Function(arg)
+    raise TypeError(f"Cannot translate {arg} to clingo Symbol")
+
 
-def atom_to_symbol(pred, args):
+def atom_to_symbol(pred, args) -> Function:
     xs = tuple(arg_to_symbol(arg) for arg in args)
-    return Function(name = pred, arguments = xs)
+    return Function(name=pred, arguments=xs)
+
 
 DEFAULT_HEURISTIC = """
 #heuristic size(N). [1000-N,true]
@@ -29,35 +36,55 @@ def atom_to_symbol(pred, args):
 program_size_at_least(M):- size(N), program_bounds(M), M <= N.
 """
 
-class Generator:
 
-    def __init__(self, settings, bkcons=[]):
+class Generator(AbstractGenerator):
+    settings: Settings
+    model: Optional[clingo.Model]
+    solver: Optional[clingo.Control]
+    handler: Optional[clingo.SolveHandle]
+
+    def __init__(self, settings: Settings, bkcons=[]):
         self.savings = 0
         self.settings = settings
         self.cached_clingo_atoms = {}
         self.handle = None
         self.pruned_sizes = set()
 
+        encoding = self.build_encoding(bkcons, settings)
+
+        with open('/tmp/ENCODING-GEN.pro', 'w') as f:
+            f.write(encoding)
+
+        solver = self.init_solver(encoding)
+        self.solver = solver
+
+        self.model = None
+
+    def init_solver(self, encoding):
+        if self.settings.single_solve:
+            solver = clingo.Control(['--heuristic=Domain', '-Wnone'])
+        solver.configuration.solve.models = 0
+        solver.add('base', [], encoding)
+        solver.ground([('base', [])])
+        return solver
+
+    def build_encoding(self, bkcons, settings):
         encoding = []
-        alan = pkg_resources.resource_string(__name__, "lp/alan.pl").decode()
+        alan = resource_string(__name__, "lp/alan.pl").decode()
         encoding.append(alan)
-
         with open(settings.bias_file) as f:
             bias_text = f.read()
-        bias_text = re.sub(r'max_vars\(\d*\).','', bias_text)
-        bias_text = re.sub(r'max_body\(\d*\).','', bias_text)
-        bias_text = re.sub(r'max_clauses\(\d*\).','', bias_text)
-
+        bias_text = re.sub(r'max_vars\(\d*\).', '', bias_text)
+        bias_text = re.sub(r'max_body\(\d*\).', '', bias_text)
+        bias_text = re.sub(r'max_clauses\(\d*\).', '', bias_text)
         # AC: NEED TO COMPLETELY REFACTOR THIS CODE
-        for p,a in settings.pointless:
+        for p, a in settings.pointless:
             bias_text = re.sub(rf'body_pred\({p},\s*{a}\)\.', '', bias_text)
             bias_text = re.sub(rf'constant\({p},.*?\).*', '', bias_text, flags=re.MULTILINE)
-
         encoding.append(bias_text)
         encoding.append(f'max_clauses({settings.max_rules}).')
         encoding.append(f'max_body({settings.max_body}).')
         encoding.append(f'max_vars({settings.max_vars}).')
-
         # ADD VARS, DIRECTIONS, AND TYPES
         head_arity = len(settings.head_literal.arguments)
         encoding.append(f'head_vars({head_arity}, {tuple(range(head_arity))}).')
@@ -68,65 +95,48 @@ def __init__(self, settings, bkcons=[]):
                 encoding.append(f'vars({arity}, {tuple(xs)}).')
                 for i, x in enumerate(xs):
                     encoding.append(f'var_pos({x}, {tuple(xs)}, {i}).')
-
         type_encoding = set()
         if self.settings.head_types:
             types = tuple(self.settings.head_types)
-            str_types = str(types).replace("'","")
+            str_types = str(types).replace("'", "")
             for i, x in enumerate(self.settings.head_types):
                 type_encoding.add(f'type_pos({str_types}, {i}, {x}).')
 
             for pred, types in self.settings.body_types.items():
                 types = tuple(types)
-                str_types = str(types).replace("'","")
+                str_types = str(types).replace("'", "")
                 for i, x in enumerate(types):
                     type_encoding.add(f'type_pos({str_types}, {i}, {x}).')
             encoding.extend(type_encoding)
-
         for pred, xs in self.settings.directions.items():
             for i, v in xs.items():
                 if v == '+':
                     encoding.append(f'direction_({pred}, {i}, in).')
                 if v == '-':
                     encoding.append(f'direction_({pred}, {i}, out).')
-
         max_size = (1 + settings.max_body) * settings.max_rules
         if settings.max_literals < max_size:
             encoding.append(f'custom_max_size({settings.max_literals}).')
-
         if settings.noisy:
             encoding.append(NOISY_ENCODING)
-
         encoding.extend(bkcons)
-
         if settings.single_solve:
             encoding.append(DEFAULT_HEURISTIC)
-
         encoding = '\n'.join(encoding)
-
-        # with open('ENCODING-GEN.pl', 'w') as f:
-            # f.write(encoding)
-
-        if self.settings.single_solve:
-            solver = clingo.Control(['--heuristic=Domain','-Wnone'])
-
-        solver.configuration.solve.models = 0
-        solver.add('base', [], encoding)
-        solver.ground([('base', [])])
-        self.solver = solver
+        return encoding
 
     def update_solver(self, size):
         # not used when learning programs without pi or recursion
         pass
 
     def get_prog(self):
         if self.handle is None:
-            self.handle = iter(self.solver.solve(yield_ = True))
+            self.handle = iter(self.solver.solve(yield_=True))
         self.model = next(self.handle, None)
         if self.model is None:
             return None
 
-        return self.parse_model_single_rule(self.model.symbols(shown = True))
+        return self.parse_model_single_rule(self.model.symbols(shown=True))
 
     def parse_model_single_rule(self, model):
         settings = self.settings
@@ -155,13 +165,13 @@ def constrain(self, tmp_new_cons):
 
             if con_type == Constraint.GENERALISATION or con_type == Constraint.BANISH:
                 con_size = None
-                if self.settings.noisy and len(xs)>2:
+                if self.settings.noisy and len(xs) > 2:
                     con_size = xs[2]
                 ground_rules2 = tuple(self.build_generalisation_constraint3(con_prog, con_size))
                 new_ground_cons.update(ground_rules2)
             elif con_type == Constraint.SPECIALISATION:
                 con_size = None
-                if self.settings.noisy and len(xs)>2:
+                if self.settings.noisy and len(xs) > 2:
                     con_size = xs[2]
                 ground_rules2 = tuple(self.build_specialisation_constraint3(con_prog, con_size))
                 new_ground_cons.update(ground_rules2)
@@ -242,12 +252,12 @@ def find_deep_bindings4(self, body):
         if len(body_types) == 0 or head_types is None:
             num_vars = len({var for atom in body for var in atom.arguments})
             for xs in permutations(range(self.settings.max_vars), num_vars):
-                x = {i:xs[i] for i in range(num_vars)}
+                x = {i: xs[i] for i in range(num_vars)}
                 yield x
             return
 
         # if there are types, only find type-safe permutations
-        var_type_lookup = {i:head_type for i, head_type in enumerate(head_types)}
+        var_type_lookup = {i: head_type for i, head_type in enumerate(head_types)}
 
         head_vars = set(range(len(self.settings.head_literal.arguments)))
         body_vars = set()
@@ -273,7 +283,7 @@ def find_deep_bindings4(self, body):
                 k = (x, y)
                 bad_type_matching.add(k)
 
-        lookup = {x:i for i, x in enumerate(body_vars)}
+        lookup = {x: i for i, x in enumerate(body_vars)}
 
         for xs in permutations(range(self.settings.max_vars), len(lookup)):
             assignment = {}
@@ -288,25 +298,27 @@ def find_deep_bindings4(self, body):
                 continue
             yield assignment
 
-def remap_variables(rule):
+
+def remap_variables(rule: Tuple[Literal,Any]):
+    head: Literal
     head, body = rule
-    head_vars = set()
+    head_vars: Set = set()
 
     if head:
-        head_vars.extend(head.arguments)
+        head_vars |= head.arguments
 
     next_var = len(head_vars)
-    lookup = {i:i for i in head_vars}
+    lookup = {i: i for i in head_vars}
 
     new_body = set()
     for pred, args in body:
         new_args = []
         for var in args:
             if var not in lookup:
                 lookup[var] = next_var
-                next_var+=1
+                next_var += 1
             new_args.append(lookup[var])
         new_atom = Literal(pred, tuple(new_args))
         new_body.add(new_atom)
 
-    return head, frozenset(new_body)
+    return head, frozenset(new_body)