Choices and Aggregates

Facts, rules, and constraints are deterministic — given them, the answer set is fixed. This page is where you hand the solver a real decision, and then count, sum, and bound what it decides.

Two constructs do the work, and they pair up. A choice rule is the generate step: it offers the solver a set of candidate atoms and lets it pick a subset. Aggregates are the test step: they count, sum, and take extremes over atoms — usually the ones a choice just generated — so you can insist the result stays within bounds. Generate a space of possibilities, then prune it: that’s the shape of most ASP programs, and the shape of this page.

We’ll build one running example — assigning a workshop’s tasks to its workers.

Choice rules

A choice rule hands the solver a decision: from a set of candidate atoms, make some subset true. Choice(element, condition=...) builds one. The element is the atom being chosen, and the condition says where the candidates come from — it’s a conditional literal, so you get one candidate per way the condition can be satisfied. Cap the decision with exactly(), at_least(), or at_most(), and attach it to a rule with when(...).derive(...):

from aspalchemy import ASPProgram, Choice, Field, Predicate, Variable

class Task(Predicate, show=False):
    name: Field[str]

class Worker(Predicate, show=False):
    name: Field[str]

class Assigned(Predicate):
    task: Field[str]
    worker: Field[str]

T, W = Variable("T"), Variable("W")

program = ASPProgram()
program.fact(*(Task(name=t) for t in ["wiring", "plumbing", "painting"]))
program.fact(*(Worker(name=w) for w in ["ada", "grace"]))

# Every task goes to exactly one worker
program.when(Task(name=T)).derive(
    Choice(Assigned(task=T, worker=W), condition=Worker(name=W)).exactly(1)
)

>>> print(program.render())
% Generated by aspalchemy ...
task("wiring").
task("plumbing").
task("painting").
worker("ada").
worker("grace").
{ assigned(T, W) : worker(W) } = 1 :- task(T).

#show.
#show assigned/2.

Read that last line as: for each task T, pick exactly one assigned atom, drawing the worker W from the workers we know about. The three cardinality methods render the way clingo spells them — exactly(1) is { ... } = 1, at_least(2).at_most(4) is 2 { ... } 4, and with no bounds at all the subset is free:

class OnCall(Predicate):
    worker: Field[str]

>>> banded = Choice(Assigned(task=T, worker=W), condition=Worker(name=W)).at_least(2).at_most(4)
>>> banded.render()
'2 { assigned(T, W) : worker(W) } 4'

A choice that needs no conditions attaches directly with choose(), and holds unconditionally:

# A bare choice: no conditions, holds unconditionally
standby = Choice(OnCall(worker=W), condition=Worker(name=W)).at_least(1)
program.choose(standby)

>>> standby.render()  # in the program, a rule of its own: '1 { on_call(W) : worker(W) }.'
'1 { on_call(W) : worker(W) }'

A single choice can offer more than one kind of atom: add(element, condition=...) appends another, and the cardinality then bounds the whole set. The bounds are checked as you build — a negative bound, or at_least(3).at_most(2), is refused right at the Python line rather than quietly becoming an unsatisfiable program:

class Approve(Predicate):
    task: Field[str]

class Defer(Predicate):
    task: Field[str]

>>> decision = Choice(Approve(task=T))
>>> decision.add(Defer(task=T))          # add() mutates, and returns nothing
>>> decision.render()
'{ approve(T); defer(T) }'

# Solve and check the choices were honored
model = program.solve().first()
for t in ["wiring", "plumbing", "painting"]:
    assert sum(1 for a in model.atoms(Assigned) if a.task == t) == 1
assert len(model.atoms(OnCall)) >= 1

And the bounds needn’t be literal integers: a Variable or Expression bound in the rule’s conditions works too, so your data can decide how many atoms to choose.

A choice is a value

A Choice is an ordinary Python value you can name, reuse, and share between rules — but its two halves, the bounds and the elements, behave differently, and the difference is worth a minute.

Bounds hand back a new choice. exactly(), at_least(), and at_most() don’t touch the choice you call them on — each returns a fresh choice carrying the bound. That’s what lets one menu be bounded two ways:

>>> menu = Choice(OnCall(worker=W), condition=Worker(name=W))
>>> weekday, weekend = menu.exactly(1), menu.exactly(2)
>>> weekday.render()
'{ on_call(W) : worker(W) } = 1'
>>> weekend.render()
'{ on_call(W) : worker(W) } = 2'
>>> menu.render()   # the original is untouched, and still reusable
'{ on_call(W) : worker(W) }'

Elements are added by mutation. add() appends to this choice and returns nothing — the same contract as list.append, so it can’t be mistaken for a value. That raises a hazard: once a rule has captured a choice, mutating it afterward would silently rewrite the rule already on record. So a captured choice freezes, and add() refuses — naming the file and line of the rule that captured it:

>>> standby.add(OnCall(worker=W))  # standby was captured by program.choose() above
Traceback (most recent call last):
  ...
RuntimeError: This Choice was captured by the rule at ... and is frozen; mutating it would silently rewrite the recorded rule. Call .copy() for a fresh, mutable Choice with the same elements, or build a new Choice.

The error names the way out. copy() returns an independent, mutable choice with the same elements; since no rule holds the copy, building on it can’t rewrite anything already recorded:

>>> extended = standby.copy()   # a fresh, mutable Choice
>>> extended.add(OnCall(worker="hot_spare"))
>>> extended.render()
'1 { on_call(W) : worker(W); on_call("hot_spare") }'
>>> standby.render()            # the captured one is untouched
'1 { on_call(W) : worker(W) }'

Bounding a frozen choice is fine for the same reason — it rewrites nothing and hands back a new choice (which is exactly why exactly() and friends are built on copy()). And freezing fences only mutation: a frozen choice is still a value, so more rules may capture and share it, rendering identically in each. That file:line in the error is part of the library’s habit that every error teaches.

class Weekend(Predicate, show=False):
    pass

rota = ASPProgram()
rota.fact(Worker(name="ada"), Weekend())
cover = Choice(OnCall(worker=W), condition=Worker(name=W)).exactly(1)
rota.choose(cover)               # captured (and frozen) here...
rota.when(Weekend()).derive(cover)  # ...and shared here, legally

>>> print(rota.render())
% Generated by aspalchemy ...
worker("ada").
weekend.
{ on_call(W) : worker(W) } = 1.
{ on_call(W) : worker(W) } = 1 :- weekend.

#show.
#show on_call/1.

Aggregates

That’s the generate step; aggregates are the test. An aggregate reduces a set of atoms to a single value you can then bound. There are five — Count, Sum, SumPlus, Min, and Max — and they all share the grammar of a choice element: an element (or tuple of terms) with an optional condition, plus add() for further elements.

from aspalchemy import Field, Predicate, PredicateArg

class Person(Predicate):
    name: Field[PredicateArg]
    age: Field[PredicateArg]

john = Person(name="john", age=30)
mary = Person(name="mary", age=25)

from aspalchemy import ANY, Count, Variable

X = Variable("X")
count = Count(X, Person(name=X, age=ANY)) > 5

Count counts distinct matching tuples. The other four reduce the first term of each tuple: Sum adds it up (SumPlus is the same, except negative weights count as zero), and Min and Max take the extremes. So a tuple’s leading term is its weight, and any trailing terms are there only to keep tuples distinct — Sum((A, X), Person(name=X, age=A)) sums one age per person, even when two people happen to share an age:

>>> from aspalchemy import Max, Sum
>>> A = Variable("A")
>>> total_age = Sum((A, X), Person(name=X, age=A))
>>> total_age.render()
'#sum{ A, X : person(X, A) }'
>>> oldest = Max(A, Person(name=ANY, age=A))
>>> oldest.render()
'#max{ A : person(_, A) }'

An aggregate never stands on its own. It becomes usable only inside a comparison — the guard — and that comparison then drops in wherever comparisons go: when() conditions, forbid(), require(). The count built above is already a full guard; Count(...) > 5 is just a comparison that happens to hold an aggregate. Handing a bare aggregate to a rule as though it were an atom is refused at construction.

census = ASPProgram()
census.fact(john, mary, Person(name="alan", age=41))
census.require(Count(X, Person(name=X, age=ANY)) >= 3)  # holds: 3 people
census.solve().first()  # would raise UnsatisfiableError if the guard failed

Guards done right

A one-sided guard is a single comparison. A band — the count has to land between two bounds — is two comparisons over the same aggregate. The rule of thumb: build the aggregate once, then compare it twice.

# One-sided: no worker carries more than two tasks
load = Count(T, Assigned(task=T, worker=W))
program.forbid(Worker(name=W), load > 2)

# A band: total assignments between 3 and 6 — two comparisons, same aggregate
class Balanced(Predicate):
    pass

workload = Count((T, W), Assigned(task=T, worker=W))
program.when(workload >= 3, workload <= 6).derive(Balanced())

>>> print(program.render())
% Generated by aspalchemy ...
task("wiring").
task("plumbing").
task("painting").
worker("ada").
worker("grace").
{ assigned(T, W) : worker(W) } = 1 :- task(T).
1 { on_call(W) : worker(W) }.
:- worker(W), #count{ T : assigned(T, W) } > 2.
balanced :- #count{ T, W : assigned(T, W) } >= 3, #count{ T, W : assigned(T, W) } <= 6.

#show.
#show assigned/2.
#show balanced/0.
#show on_call/1.
>>> model = program.solve().first()
>>> bool(model.atoms(Balanced))  # 3 tasks, one worker each: the band holds
True

The render repeats the aggregate on both sides of the band, which looks wasteful. It isn’t: gringo (clingo’s grounder — the half that instantiates your rules over the data) recognizes the shared aggregate and grounds it once, so the two comparisons cost no more than clingo’s native interval literal (3 <= #count{...} <= 6, a spelling ASPAlchemy deliberately doesn’t model).

What you do want to avoid is bind-then-compare: binding the count to a variable — N == Count(...) — and then testing N against the bounds. It reads like one aggregate, but it forces gringo to ground a fresh aggregate for every feasible value of N, with the band’s width as the multiplier. Keep the variable out of it: compare the aggregate directly, once per bound. (A first-class banded form is on the wishlist; until then, two comparisons is the idiom, and it costs nothing.)

Cardinality tests are not choices

One last place the choice syntax can trip you. In clingo, braces in what a rule derives mean choose — but the very same braces among a rule’s conditions, 2 { p(X) } 4 :- q., mean something completely different: a cardinality test, counting how many p(X) are true. Nothing is chosen. (That’s the head-versus-body split from Statements and Terms; the error below names it in clingo’s own terms.) So ASPAlchemy refuses a Choice used as a condition, and the message points at the right spelling:

>>> program.forbid(Choice(OnCall(worker=W), condition=Worker(name=W)).at_least(3))
Traceback (most recent call last):
  ...
ValueError: A Choice belongs in a rule head, where braces CHOOSE. In a body, clingo's braces mean a cardinality TEST — a different construct aspalchemy spells as a Count comparison: Count(X, condition=...) >= n.

A cardinality test is a Count guard — which is exactly what it means — and it grounds identically to the brace form:

# "At most two workers on call" — a cardinality test, spelled as one
program.require(Count(W, OnCall(worker=W)) <= 2)
model = program.solve().first()
assert len(model.atoms(OnCall)) <= 2

A two-sided test (2 { p(X) } 4 among conditions) is just the band idiom from above: two comparisons over one Count. Keeping the condition-brace syntax out is a deliberate strictness — one syntax should mean one thing, and here braces mean choose, which is why they belong only in what a rule derives.