"""General attribution functions (Falmagne & Doignon 2011, §5.1–5.2).
Unlike a surmise function, an attribution need not contain its item in
each clause, satisfy refinement, or have incomparable clauses. Every item
must still have a nonempty family of clauses; an empty clause is allowed.
"""
from __future__ import annotations
from collections.abc import Collection, Mapping
from heapq import heappop, heappush
from typing import TYPE_CHECKING
from knowledgespaces.structures.surmise_function import SurmiseFunction
if TYPE_CHECKING:
from knowledgespaces.structures.knowledge_structure import KnowledgeStructure
[docs]
class Attribution:
"""An item-indexed family of alternative prerequisite sets.
``clauses`` must cover the nonempty ``domain`` exactly. Duplicates are
removed, but redundant clauses are retained in this representation.
An absent prerequisite is expressed by the family ``[[]]``, not ``[]``.
"""
__slots__ = ("_clauses", "_domain")
def __init__(
self, domain: Collection[str], clauses: Mapping[str, Collection[Collection[str]]]
) -> None:
self._domain = frozenset(domain)
if not self._domain or set(clauses) != self._domain:
raise ValueError("Attribution clauses must cover exactly a nonempty domain.")
parsed = {q: frozenset(frozenset(c) for c in clauses[q]) for q in self._domain}
if any(not family for family in parsed.values()):
raise ValueError(
"Each item needs a nonempty clause family (an empty clause is allowed)."
)
if any(not c <= self._domain for family in parsed.values() for c in family):
raise ValueError("Clauses contain items outside the attribution domain.")
self._clauses = parsed
@property
def domain(self) -> frozenset[str]:
return self._domain
[docs]
def clauses_for(self, item: str) -> frozenset[frozenset[str]]:
"""The original clause family, including any redundant clauses."""
if item not in self._domain:
raise ValueError("Unknown attribution item.")
return self._clauses[item]
def __contains__(self, state: Collection[str]) -> bool:
"""Whether the set is a state defined by Equation (5.2).
Each item in the set must have some clause included in that set.
This test requires neither canonicalization nor state enumeration.
"""
state = frozenset(state)
return state <= self._domain and all(
any(c <= state for c in self._clauses[q]) for q in state
)
@property
def is_surmise_function(self) -> bool:
"""Whether the original clauses already satisfy all surmise axioms."""
try:
SurmiseFunction(self._domain, self._clauses)
except ValueError:
return False
return True
[docs]
def to_surmise_function(self, *, max_candidates: int = 100_000) -> SurmiseFunction:
"""Canonical surmise function defining exactly the same knowledge space.
Search for all inclusion-minimal states containing each item. Start
from its clauses with the item added; whenever an included item lacks
a satisfied clause, branch over that item's alternatives. Sets grow
monotonically. A valid state prunes its supersets from further work.
This works with cycles and does not mistake a clause union for a
state before every included item's requirements are satisfied.
``max_candidates`` bounds distinct sets visited in each item-rooted
search. Exceeding it raises instead of returning a partial function.
The worst case is exponential, but a powerset of states is not
materialized as an intermediate representation.
"""
_positive_limit(max_candidates, "max_candidates")
canonical = {
root: self._rooted_atoms(root, max_candidates) for root in sorted(self._domain)
}
return SurmiseFunction(self._domain, canonical)
def _rooted_atoms(self, root: str, max_candidates: int) -> frozenset[frozenset[str]]:
seen: set[frozenset[str]] = set()
pending: list[tuple[int, tuple[str, ...]]] = []
atoms: set[frozenset[str]] = set()
def enqueue(state: frozenset[str]) -> None:
if state in seen or any(a <= state for a in atoms):
return
seen.add(state)
if len(seen) > max_candidates:
raise ValueError(f"Attribution search for {root!r} exceeds max_candidates.")
heappush(pending, (len(state), tuple(sorted(state))))
for clause in self._clauses[root]:
enqueue(clause | {root})
while pending:
_, labels = heappop(pending)
state = frozenset(labels)
if any(a <= state for a in atoms):
continue
unsatisfied = next(
(q for q in labels if not any(c <= state for c in self._clauses[q])), None
)
if unsatisfied is None:
atoms.add(state)
else:
for clause in self._clauses[unsatisfied]:
enqueue(state | clause)
return frozenset(atoms)
[docs]
def to_knowledge_space(
self, *, max_candidates: int = 100_000, max_items: int | None = None
) -> KnowledgeStructure:
"""Canonicalize and enumerate the union span, with both resource guards."""
return self.to_surmise_function(max_candidates=max_candidates).to_knowledge_space(
max_items=max_items
)
[docs]
def expand(self, domain: Collection[str]) -> Attribution:
"""Extend to a larger domain with no restrictions on the added items.
Original clauses are unchanged; new items have the empty prerequisite
clause. Call ``to_surmise_function`` for the canonical expansion.
"""
domain = frozenset(domain)
if not self._domain <= domain:
raise ValueError("Expanded domain must contain every original item.")
return Attribution(
domain, {q: self._clauses[q] if q in self._domain else [frozenset()] for q in domain}
)
[docs]
@classmethod
def from_relation(
cls, domain: Collection[str], pairs: Collection[tuple[str, str]]
) -> Attribution:
"""Cast any binary relation as an attribution (Definition 5.1.4).
Pair (a,b) means a is required for b. No reflexive or transitive
closure is added. An item with no incoming pairs has an empty clause.
"""
domain = frozenset(domain)
required: dict[str, set[str]] = {q: set() for q in domain}
for a, b in pairs:
if a not in domain or b not in domain:
raise ValueError("Relation pair contains an item outside the domain.")
required[b].add(a)
return cls(domain, {q: [required[q]] for q in domain})
[docs]
@classmethod
def from_surmise_function(cls, function: SurmiseFunction) -> Attribution:
"""Preserve every clause of an already canonical surmise function."""
return cls(function.domain, {q: function.clauses_for(q) for q in function.domain})
def __eq__(self, other: object) -> bool:
if not isinstance(other, Attribution):
return NotImplemented
return self._domain == other._domain and self._clauses == other._clauses
def __repr__(self) -> str:
return f"Attribution(n_items={len(self._domain)}, n_clauses={sum(map(len, self._clauses.values()))})"
def _positive_limit(value: int, name: str) -> None:
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
raise ValueError(f"{name} must be a positive integer.")