pynmms.rdf.closure

pynmms.rdf.closure

Semi-naive closure of Horn rules over triples.

Two uses:

• :meth:ClosureEngine.close -- full closure of a small graph (the in-memory backend's regime materialisation, and the test oracle's counterpart).
• :meth:ClosureEngine.extend -- the per-node step of a regime base: given a store that already holds cl_R(G) and a small set of extra triples (the atoms proof rules have moved into Γ), derive cl_R(G ∪ extras) \ cl_R(G) by firing only rules with at least one premise matching a new triple. Remaining premises are joined against the store's closure and the new set. Cost is proportional to the new triples and their joins, never to |G|.

Rules are range-restricted (def:entailmentregime), so every conclusion is ground.

Join = Callable[[list[Pattern], Bindings], Iterable[Bindings]] module-attribute

Answer several patterns at once against the store: all extensions of the given bindings that make every pattern a store triple. One round trip on a remote backend.

ClosureEngine

Fires a regime's rules semi-naively.

Source code in src/pynmms/rdf/closure.py

class ClosureEngine:
    """Fires a regime's rules semi-naively."""

    def __init__(self, regime: Regime) -> None:
        self.regime = regime
        # (rule, premise index) pairs keyed by the premise's constant predicate,
        # with wildcard-predicate premises under None.
        self._by_pred: dict[Node | None, list[tuple[AnyRule, int]]] = {}
        for rule in regime.rules:
            if isinstance(rule, ProceduralRule):
                for trig in rule.triggers:
                    self._by_pred.setdefault(trig, []).append((rule, -1))
                continue
            for i, prem in enumerate(rule.premises):
                key = None if isinstance(prem[1], Var) else prem[1]
                self._by_pred.setdefault(key, []).append((rule, i))

    def _candidates(self, t: Triple) -> Iterable[tuple[AnyRule, int]]:
        yield from self._by_pred.get(t[1], ())
        yield from self._by_pred.get(None, ())

    def _fire(self, rule: AnyRule, i: int, t: Triple, lookup: Lookup) -> Iterator[Triple | None]:
        """All conclusions of *rule* with premise *i* matched to *t*."""
        if isinstance(rule, ProceduralRule):
            yield from rule.fire(t, lookup)
            return
        b0 = unify(rule.premises[i], t, {})
        if b0 is None:
            return
        rest = [p for j, p in enumerate(rule.premises) if j != i]

        def join(k: int, b: Bindings) -> Iterator[Bindings]:
            if k == len(rest):
                yield b
                return
            for cand in lookup(instantiate(rest[k], b)):
                b2 = unify(rest[k], cand, b)
                if b2 is not None:
                    yield from join(k + 1, b2)

        yield from self._conclude(rule, join(0, b0))

    def _fire_batched(
        self, rule: Rule, i: int, t: Triple, new: _TripleIndex, store_join: Join
    ) -> Iterator[Triple | None]:
        """Like ``_fire`` but with the store-resident premises answered in one query.

        The remaining premises are split into those matched against the
        in-process ``new`` set and those handed to the store. Every subset is
        tried; the ``new`` positions are bound first (they are few) and the
        store is then asked for all extensions over the other positions at
        once. ``new`` and the store are disjoint, so the subsets partition the
        space and nothing is derived twice.
        """
        b0 = unify(rule.premises[i], t, {})
        if b0 is None:
            return
        rest = [p for j, p in enumerate(rule.premises) if j != i]
        n = len(rest)

        def bindings() -> Iterator[Bindings]:
            for mask in range(1 << n):
                from_new = [j for j in range(n) if mask >> j & 1]
                from_store = [rest[j] for j in range(n) if not mask >> j & 1]

                def go(idx: int, b: Bindings) -> Iterator[Bindings]:
                    if idx == len(from_new):
                        if from_store:
                            yield from store_join(from_store, b)
                        else:
                            yield b
                        return
                    pat = rest[from_new[idx]]
                    for cand in new.match(instantiate(pat, b)):
                        b2 = unify(pat, cand, b)
                        if b2 is not None:
                            yield from go(idx + 1, b2)

                yield from go(0, b0)

        yield from self._conclude(rule, bindings())

    @staticmethod
    def _conclude(rule: Rule, bindings: Iterable[Bindings]) -> Iterator[Triple | None]:
        for b in bindings:
            if rule.guard is not None and not rule.guard(b):
                continue
            yield None if rule.conclusion is None else ground(rule.conclusion, b)

    def extend(
        self,
        extras: Iterable[Triple],
        store_lookup: Lookup | None,
        store_contains: Callable[[Triple], bool] | None,
        store_join: Join | None = None,
    ) -> tuple[_TripleIndex, bool]:
        """Close ``store ∪ extras`` incrementally.

        Returns the set of triples in the closure that are *not* in the store
        (the extras themselves included) and whether ⊥ was derived. With
        *store_join* the pattern rules batch their store-side premises into
        one query per firing (procedural rules still use *store_lookup*).
        """
        new = _TripleIndex()
        frontier: list[Triple] = []
        for t in extras:
            if store_contains is not None and store_contains(t):
                continue
            if new.add(t):
                frontier.append(t)

        def lookup(pattern: tuple[Node | None, Node | None, Node | None]) -> Iterator[Triple]:
            if store_lookup is not None:
                yield from store_lookup(pattern)
            yield from new.match(pattern)

        bottom = False
        rounds = 0
        while frontier:
            rounds += 1
            next_frontier: list[Triple] = []
            for t in frontier:
                for rule, i in self._candidates(t):
                    if store_join is not None and isinstance(rule, Rule):
                        conclusions = self._fire_batched(rule, i, t, new, store_join)
                    else:
                        conclusions = self._fire(rule, i, t, lookup)
                    for concl in conclusions:
                        if concl is None:
                            if not bottom:
                                logger.debug("closure: rule %s derives false", rule.name)
                            bottom = True
                            continue
                        if concl in new or (store_contains is not None and store_contains(concl)):
                            continue
                        new.add(concl)
                        next_frontier.append(concl)
            frontier = next_frontier
        logger.debug(
            "closure extend: %d new triples in %d rounds (bottom=%s)", len(new), rounds, bottom
        )
        return new, bottom

    def close(self, triples: Iterable[Triple]) -> tuple[set[Triple], bool]:
        """Full closure of *triples* under the regime, axioms included."""
        new, bottom = self.extend(
            (*self.regime.axioms, *triples), store_lookup=None, store_contains=None
        )
        return new.triples, bottom

close(triples)

Full closure of triples under the regime, axioms included.

Source code in src/pynmms/rdf/closure.py

def close(self, triples: Iterable[Triple]) -> tuple[set[Triple], bool]:
    """Full closure of *triples* under the regime, axioms included."""
    new, bottom = self.extend(
        (*self.regime.axioms, *triples), store_lookup=None, store_contains=None
    )
    return new.triples, bottom

extend(extras, store_lookup, store_contains, store_join=None)

Close store ∪ extras incrementally.

Returns the set of triples in the closure that are not in the store (the extras themselves included) and whether ⊥ was derived. With store_join the pattern rules batch their store-side premises into one query per firing (procedural rules still use store_lookup).

Source code in src/pynmms/rdf/closure.py

def extend(
    self,
    extras: Iterable[Triple],
    store_lookup: Lookup | None,
    store_contains: Callable[[Triple], bool] | None,
    store_join: Join | None = None,
) -> tuple[_TripleIndex, bool]:
    """Close ``store ∪ extras`` incrementally.

    Returns the set of triples in the closure that are *not* in the store
    (the extras themselves included) and whether ⊥ was derived. With
    *store_join* the pattern rules batch their store-side premises into
    one query per firing (procedural rules still use *store_lookup*).
    """
    new = _TripleIndex()
    frontier: list[Triple] = []
    for t in extras:
        if store_contains is not None and store_contains(t):
            continue
        if new.add(t):
            frontier.append(t)

    def lookup(pattern: tuple[Node | None, Node | None, Node | None]) -> Iterator[Triple]:
        if store_lookup is not None:
            yield from store_lookup(pattern)
        yield from new.match(pattern)

    bottom = False
    rounds = 0
    while frontier:
        rounds += 1
        next_frontier: list[Triple] = []
        for t in frontier:
            for rule, i in self._candidates(t):
                if store_join is not None and isinstance(rule, Rule):
                    conclusions = self._fire_batched(rule, i, t, new, store_join)
                else:
                    conclusions = self._fire(rule, i, t, lookup)
                for concl in conclusions:
                    if concl is None:
                        if not bottom:
                            logger.debug("closure: rule %s derives false", rule.name)
                        bottom = True
                        continue
                    if concl in new or (store_contains is not None and store_contains(concl)):
                        continue
                    new.add(concl)
                    next_frontier.append(concl)
        frontier = next_frontier
    logger.debug(
        "closure extend: %d new triples in %d rounds (bottom=%s)", len(new), rounds, bottom
    )
    return new, bottom

apply(pattern, b)

Substitute b into pattern; unbound variables stay variables.

The result is loosely typed: callers narrow it with :func:is_ground before treating it as a triple.

Source code in src/pynmms/rdf/closure.py

def apply(pattern: Pattern, b: Bindings) -> tuple[Any, Any, Any]:
    """Substitute *b* into *pattern*; unbound variables stay variables.

    The result is loosely typed: callers narrow it with :func:`is_ground`
    before treating it as a triple.
    """
    return tuple(b.get(x, x) if isinstance(x, Var) else x for x in pattern)  # type: ignore[return-value]

ground(pattern, b)

Substitute b into a pattern every variable of which b binds.

Source code in src/pynmms/rdf/closure.py

def ground(pattern: Pattern, b: Bindings) -> Triple:
    """Substitute *b* into a pattern every variable of which *b* binds."""
    t = instantiate(pattern, b)
    assert None not in t, "range restriction guarantees ground conclusions"
    return t  # type: ignore[return-value]

instantiate(pattern, b)

Substitute b into pattern as a lookup key; unbound variables become None.

Source code in src/pynmms/rdf/closure.py

def instantiate(pattern: Pattern, b: Bindings) -> tuple[Node | None, Node | None, Node | None]:
    """Substitute *b* into *pattern* as a lookup key; unbound variables become ``None``."""
    out: list[Node | None] = []
    for term in pattern:
        out.append(b.get(term) if isinstance(term, Var) else term)  # type: ignore[arg-type]
    return (out[0], out[1], out[2])

is_ground(pattern)

Does pattern have no variables?

Source code in src/pynmms/rdf/closure.py

def is_ground(pattern: Pattern) -> bool:
    """Does *pattern* have no variables?"""
    return not any(isinstance(x, Var) for x in pattern)

join_patterns(patterns, bindings, lookup)

All extensions of bindings making every pattern a triple of lookup.

Source code in src/pynmms/rdf/closure.py

def join_patterns(
    patterns: list[Pattern], bindings: Bindings, lookup: Lookup
) -> Iterator[Bindings]:
    """All extensions of *bindings* making every pattern a triple of *lookup*."""
    patterns = order_patterns(patterns, bindings)

    def go(k: int, b: Bindings) -> Iterator[Bindings]:
        if k == len(patterns):
            yield b
            return
        for cand in lookup(instantiate(patterns[k], b)):
            b2 = unify(patterns[k], cand, b)
            if b2 is not None:
                yield from go(k + 1, b2)

    return go(0, bindings)

match_patterns(patterns, lookup)

Find one instance mapping making every pattern a triple in lookup.

This is the witness search of lem:witnesschar: with the blank nodes of a succedent graph H as variables, H is entailed iff some binding puts all of μ(H) in the closure. Returns the binding or None.

Source code in src/pynmms/rdf/closure.py

def match_patterns(patterns: Iterable[Pattern], lookup: Lookup) -> Bindings | None:
    """Find one instance mapping making every pattern a triple in *lookup*.

    This is the witness search of ``lem:witnesschar``: with the blank nodes of a
    succedent graph H as variables, ``H`` is entailed iff some binding puts
    all of ``μ(H)`` in the closure. Returns the binding or ``None``.
    """
    pats = order_patterns(list(patterns), {})

    def go(k: int, b: Bindings) -> Bindings | None:
        if k == len(pats):
            return b
        for cand in lookup(instantiate(pats[k], b)):
            b2 = unify(pats[k], cand, b)
            if b2 is not None:
                found = go(k + 1, b2)
                if found is not None:
                    return found
        return None

    return go(0, {})

order_patterns(patterns, bindings)

Greedy join order: at each step the pattern with the most bound terms.

A pattern query such as ?b a C . ?b name "x" must start from the name, which matches once, not from the class, which matches every member.

Source code in src/pynmms/rdf/closure.py

def order_patterns(patterns: list[Pattern], bindings: Bindings) -> list[Pattern]:
    """Greedy join order: at each step the pattern with the most bound terms.

    A pattern query such as ``?b a C . ?b name "x"`` must start from the
    name, which matches once, not from the class, which matches every member.
    """
    remaining = list(patterns)
    bound = set(bindings)
    ordered: list[Pattern] = []
    while remaining:
        best = max(remaining, key=lambda p: _selectivity(p, bound))
        remaining.remove(best)
        ordered.append(best)
        bound |= {t for t in best if isinstance(t, Var)}
    return ordered

unify(pattern, t, bindings=None)

Extend bindings so that pattern matches the ground triple t, or None.

A variable already bound to a different term fails the match; the input bindings are not modified.

Source code in src/pynmms/rdf/closure.py

def unify(pattern: Pattern, t: Triple, bindings: Bindings | None = None) -> Bindings | None:
    """Extend *bindings* so that *pattern* matches the ground triple *t*, or ``None``.

    A variable already bound to a different term fails the match; the input
    bindings are not modified.
    """
    b: Bindings = dict(bindings) if bindings else {}
    for term, value in zip(pattern, t):
        if isinstance(term, Var):
            bound = b.get(term)
            if bound is None:
                b[term] = value
            elif bound != value:
                return None
        elif term != value:
            return None
    return b