Skip to content

Commit

Permalink
Merge pull request #124 from potassco/missing-methods-clormclingocontrol
Browse files Browse the repository at this point in the history
Added the on_unsat and on_core callbacks for clorm.clingo.Control
  • Loading branch information
daveraja authored Sep 20, 2023
2 parents 6880419 + 7f1acb4 commit ed70d51
Show file tree
Hide file tree
Showing 2 changed files with 90 additions and 13 deletions.
25 changes: 14 additions & 11 deletions clorm/_clingo.py
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,9 @@ def facts(self, *args: Any, **kwargs: Any) -> FactBase:
atoms: select all atoms in the model (Default: False)
terms: select all terms displayed with #show statements (Default: False)
shown: select all atoms and terms (Default: False)
theory: select atoms added with Model.extend() (Default: False)
complement: return the complement of the answer set w.r.t. to the
atoms known to the grounder. (Default: False)
raise_on_empty: raise a ValueError if the resulting FactBase is empty
(Default: False)
Expand All @@ -123,12 +126,12 @@ def facts(self, *args: Any, **kwargs: Any) -> FactBase:
nkwargs = dict(kwargs)
if len(nargs) >= 1 and "unifier" in nkwargs:
raise TypeError("facts() got multiple values for argument 'unifier'")
if len(nargs) >= 5 and "raise_on_empty" in nkwargs:
if len(nargs) >= 7 and "raise_on_empty" in nkwargs:
raise TypeError("facts() got multiple values for argument 'raise_on_empty'")

raise_on_empty = nkwargs.pop("raise_on_empty", False)
if len(nargs) >= 5:
raise_on_empty = nargs.pop(4)
if len(nargs) >= 7:
raise_on_empty = nargs.pop(6)
unifier = nkwargs.pop("unifier", None)
if len(nargs) >= 1:
unifier = nargs.pop(0)
Expand Down Expand Up @@ -464,14 +467,14 @@ def solve(self, *args: Any, **kwargs: Any) -> Union[ClormSolveHandle, oclingo.So
# keyword for Python 3.7+.
async_keyword = "async_" if oclingo.__version__ > "5.3.1" else "async"

posnargs = [
"assumptions",
"on_model",
"on_statistics",
"on_finish",
"yield_",
async_keyword,
]
posnargs = ["assumptions", "on_model"]

if oclingo.__version__ >= "5.5.0":
posnargs.append("on_unsat")
posnargs.extend(["on_statistics", "on_finish"])
if oclingo.__version__ >= "5.5.0":
posnargs.append("on_core")
posnargs.extend(["yield_", async_keyword])
validargs = set(posnargs)

# translate all positional arguments into keyword arguments.
Expand Down
78 changes: 76 additions & 2 deletions tests/test_clingo.py
Original file line number Diff line number Diff line change
Expand Up @@ -701,7 +701,7 @@ class Af(Predicate):
check_errmsg("facts() got multiple values for argument 'unifier'", ctx)

with self.assertRaises(TypeError) as ctx:
fb = m.facts([Af], True, True, True, True, raise_on_empty=True)
fb = m.facts([Af], True, True, True, True, True, True, raise_on_empty=True)
check_errmsg("facts() got multiple values for argument 'raise_on_empty'", ctx)

# --------------------------------------------------------------------------
Expand Down Expand Up @@ -984,7 +984,10 @@ def on_statistics(stp, acc):
# Calling using positional arguments
om_called = False
os_called = False
sr = ctrl.solve([], on_model, on_statistics)
if oclingo.__version__ < "5.5.0":
sr = ctrl.solve([], on_model, on_statistics)
else:
sr = ctrl.solve([], on_model, None, on_statistics)
self.assertTrue(om_called)
self.assertTrue(os_called)

Expand All @@ -995,6 +998,77 @@ def on_statistics(stp, acc):
self.assertFalse(om_called)
self.assertTrue(os_called)

# --------------------------------------------------------------------------
# Test the solve with on_unsat
# --------------------------------------------------------------------------
def test_solve_with_on_unsat(self):
# There is no on_unsat or on_core functions for older versions.
if oclingo.__version__ < "5.5.0":
return

# Program to test optimisation for on_unsat callback - note: setting the appropriate
# optimisation strategy is important to trigger the callback.
prgstr = """
1 { p(X); q(X) } 1 :- X=1..3.
#minimize { 1,p,X: p(X); 1,q,X: q(X) }.
"""

ctrl = cclingo.Control()
conf = ctrl.configuration
conf.solver.opt_strategy = "usc,oll,0"
add_program_string(ctrl, prgstr)
ctrl.ground([("base", [])])

def on_unsat(ints):
nonlocal ou_called
ou_called = True

# Calling using positional arguments
ou_called = False
sr = ctrl.solve([], None, on_unsat)
self.assertTrue(ou_called)

# Caling using keyword arguments
ou_called = False
sr = ctrl.solve(on_unsat=on_unsat)
self.assertTrue(ou_called)

# --------------------------------------------------------------------------
# Test the solve with on_core
# --------------------------------------------------------------------------
def test_solve_with_on_core(self):
# There is no on_unsat or on_core functions for older versions.
if oclingo.__version__ < "5.5.0":
return

class F(Predicate):
num1 = IntegerField()

# Program to test unsatisfiable cores
prgstr = """
2 {f(1..3)} 2.
"""
all_facts = FactBase([F(1), F(2), F(3)])
ctrl = cclingo.Control(unifier=[F])
add_program_string(ctrl, prgstr)
ctrl.ground([("base", [])])
all_literals = [ctrl.symbolic_atoms[f.raw].literal for f in all_facts]

def on_core(ints):
nonlocal oc_called
assert set(all_literals) == set(ints)
oc_called = True

# Calling using positional arguments
oc_called = False #
sr = ctrl.solve([(all_facts, True)], None, None, None, None, on_core)
self.assertTrue(oc_called)

# Calling using keyword arguments
oc_called = False
sr = ctrl.solve(assumptions=[(all_facts, True)], on_core=on_core)
self.assertTrue(oc_called)

# --------------------------------------------------------------------------
# Test accessing some other control variables that are not explcitly wrapped
# --------------------------------------------------------------------------
Expand Down

0 comments on commit ed70d51

Please sign in to comment.