"""
Block 2+ of the QUERY algorithm: learning space refinement.
Refines an ordinal space (from Block 1) into a learning space by
querying group prerequisites. Block N tests antecedent sets of size N.
Block 2 handles antecedent size 2, Block 3 size 3, etc. The algorithm
is identical at every level --- only the antecedent cardinality changes.
Two inference mechanisms reduce expert queries:
- Negative monotonicity: if q is globally minimal (Qmax), then (A,q)=NO.
- Positive monotonicity: if any subset of A already implies q, then (A,q)=YES.
Applying a positive answer removes the states contradicting it
(Definition 16.1.5), which is admissible only when the removal leaves a
learning space (Theorem 16.1.6). Two treatments of an inadmissible
("blocked") positive are implemented, selectable via ``algorithm``:
``"pending"`` (default)
The pending-table mechanism of Falmagne & Doignon (2011),
Section 16.1 (Example 16.1.13), applied at the state level: a
blocked positive is buffered with pending status and re-tested
after every successful removal, within and across blocks, until the
collection of states stabilizes. By Theorem 16.1.16 the procedure
cannot get jammed: with an error-free expert whose latent structure
is a learning space, and queries of every antecedent size, the
stabilized output is the latent space. Section 16.2 of the book
reformulates the same mechanism on the surmise function
(Algorithm 16.2.11) to avoid storing the state collection --- a
memory concern (Drawback 16.1.12(b)) that does not bind at this
package's documented domain ceiling, so the state-level formulation
is used here.
``"one_pass"``
The naive one-pass scheme of Algorithm 16.1.10 (cf. 15.1.1): a
positive whose removal is blocked is recorded as negative on
structural grounds, without consulting the expert, and never
revisited. The book documents the defect (Example 16.1.12(a)): a
removal blocked at one point can become admissible after later
removals, so a truthful positive can be suppressed permanently and
the result can strictly contain the latent space and depend on the
item order. Retained for comparison and reproducibility.
References:
Falmagne, J.-C., & Doignon, J.-P. (2011).
Learning Spaces, Chapter 16. Springer-Verlag.
"""
from __future__ import annotations
from collections.abc import Collection
from dataclasses import dataclass, field
from itertools import combinations
from typing import Literal
from knowledgespaces._limits import DEFAULT_MAX_N_ITEMS, check_domain_size
from knowledgespaces.query.expert import Expert
from knowledgespaces.query.types import InferenceSource, Query, QueryAnswer
from knowledgespaces.structures.knowledge_structure import KnowledgeStructure
Algorithm = Literal["pending", "one_pass"]
"""Treatment of blocked positive answers; see the module docstring."""
PendingQuery = tuple[frozenset[str], str]
"""A positive query (antecedent, consequent) awaiting an admissible removal."""
[docs]
@dataclass
class BlockNStats:
"""Statistics from a Block N execution."""
expert_queries: int = 0
inferred_negative_monotonicity: int = 0
inferred_positive_monotonicity: int = 0
inferred_structural: int = 0
pending_added: int = 0
pending_resolved: int = 0
@property
def total_inferences(self) -> int:
return (
self.inferred_negative_monotonicity
+ self.inferred_positive_monotonicity
+ self.inferred_structural
)
@property
def total_candidates(self) -> int:
return self.expert_queries + self.total_inferences
[docs]
@dataclass
class BlockNResult:
"""Result of a Block N query phase.
``pending`` holds the positive answers (from this block or carried
in from earlier ones) whose state removal is still blocked when the
block ends; it is empty under ``algorithm="one_pass"``.
"""
structure: KnowledgeStructure
positive: set[tuple[frozenset[str], str]]
negative: set[tuple[frozenset[str], str]]
stats: BlockNStats
log: list[QueryAnswer] = field(default_factory=list)
pending: tuple[PendingQuery, ...] = ()
def _is_learning_space(states: set[frozenset[str]], domain: frozenset[str]) -> bool:
"""Check that no state is hanging (cf. Observation 16.1.4).
This verifies the necessary condition for a learning space used by
the QUERY algorithm's admissibility test (cf. Theorem 16.1.6): the
structure contains ∅ and Q, and every non-empty state has at least
one single-item removal that yields another state.
For knowledge spaces (closed under union), this local condition is
equivalent to the full learning space definition. During Block N
execution, L_current always starts as a learning space and is only
modified by removing sets D(A, q), which preserves union closure
(Theorem 16.1.6), so this check is sufficient.
"""
if frozenset() not in states or domain not in states:
return False
for state in states:
if not state:
continue
# A state is hanging if no single-item removal yields another state
if not any(state - {q} in states for q in state):
return False
return True
def _try_apply(
L: set[frozenset[str]],
A: frozenset[str],
q: str,
domain: frozenset[str],
) -> tuple[set[frozenset[str]], Literal["vacuous", "applied", "blocked"]]:
"""Attempt to implement a positive answer to (A, q) on ``L``.
Returns the (possibly reduced) collection and a status:
``"vacuous"`` when no state contradicts the answer (D = ∅),
``"applied"`` when D was removed and the result is a learning space,
``"blocked"`` when the removal would create a hanging state
(Theorem 16.1.6) and ``L`` is returned unchanged.
"""
D = {K for K in L if q in K and A.isdisjoint(K)}
if not D:
return L, "vacuous"
L_potential = L - D
if _is_learning_space(L_potential, domain):
return L_potential, "applied"
return L, "blocked"
def _drain_pending(
L: set[frozenset[str]],
pending: list[PendingQuery],
domain: frozenset[str],
stats: BlockNStats,
) -> tuple[set[frozenset[str]], list[PendingQuery]]:
"""Re-test pending positives until the collection of states stabilizes.
Implements the systematic revisiting of Falmagne & Doignon (2011),
Section 16.1: every successful removal can unblock earlier positives,
so passes repeat until one completes without a removal. Entries whose
removal set has become empty are satisfied and dropped.
"""
progress = True
while progress:
progress = False
remaining: list[PendingQuery] = []
for A, q in pending:
L_next, status = _try_apply(L, A, q, domain)
if status == "blocked":
remaining.append((A, q))
else:
stats.pending_resolved += 1
if status == "applied":
L = L_next
progress = True
pending = remaining
return L, pending
[docs]
def run_block_n(
items: list[str],
structure: KnowledgeStructure,
prior_positive: set[tuple[frozenset[str], str]],
expert: Expert,
*,
antecedent_size: int = 2,
minimal_global: frozenset[str] = frozenset(),
max_items: int | None = None,
algorithm: Algorithm = "pending",
pending_in: Collection[PendingQuery] | None = None,
) -> BlockNResult:
"""Execute Block N of the QUERY algorithm.
Parameters
----------
items : list[str]
The domain of items.
structure : KnowledgeStructure
The current knowledge structure to refine.
prior_positive : set[tuple[frozenset[str], str]]
Positive relations from all previous blocks. Each entry is
(antecedent_set, consequent). Single-item antecedents come
from Block 1's closure: {(frozenset({a}), b) for (a,b) in closure}.
expert : Expert
The expert to query.
antecedent_size : int
Size of antecedent sets to test. Default 2 (standard Block 2).
minimal_global : frozenset[str]
Items certified globally minimal by Qmax in Block 1.
max_items : int | None
Hard limit override for the ``|Q|`` preflight check. Block N scales
as ``O(|Q| * C(|Q|-1, k))``; values above ~25 are not practical.
algorithm : {"pending", "one_pass"}
Treatment of positive answers whose removal is blocked; see the
module docstring. The default ``"pending"`` buffers them and
re-tests after every removal (Section 16.1, Theorem 16.1.16);
``"one_pass"`` records them as structural negatives without
consulting the expert (Algorithm 16.1.10).
pending_in : Collection[PendingQuery] | None
Blocked positives carried over from earlier blocks (``"pending"``
only). They are re-tested against the current structure before
and during this block.
Returns
-------
BlockNResult
Refined structure, discovered relations, stats, and (under
``"pending"``) the positives still blocked at the end.
"""
check_domain_size(
len(items),
max_threshold=max_items if max_items is not None else DEFAULT_MAX_N_ITEMS,
context="run_block_n",
)
if algorithm not in ("pending", "one_pass"):
raise ValueError(f"algorithm must be 'pending' or 'one_pass', got {algorithm!r}")
domain = frozenset(items)
L_current: set[frozenset[str]] = set(structure.states)
P_current: set[tuple[frozenset[str], str]] = set(prior_positive)
N_current: set[tuple[frozenset[str], str]] = set()
stats = BlockNStats()
log: list[QueryAnswer] = []
# pending_in is a "pending"-only concept: the one-pass scheme never
# buffers, so its result documents an always-empty pending tuple.
pending: list[PendingQuery] = list(pending_in or []) if algorithm == "pending" else []
# Carried-in positives may already be admissible on this structure.
if pending:
L_current, pending = _drain_pending(L_current, pending, domain, stats)
def _handle_positive(A: frozenset[str], q: str) -> None:
"""Implement a (recorded or inferred) positive answer."""
nonlocal L_current, pending
L_next, status = _try_apply(L_current, A, q, domain)
if status == "blocked":
pending.append((A, q))
stats.pending_added += 1
elif status == "applied":
L_current = L_next
L_current, pending = _drain_pending(L_current, pending, domain, stats)
# Generate candidate queries: all (A, q) with |A| = antecedent_size
candidates: list[tuple[frozenset[str], str]] = []
for q in items:
others = [x for x in items if x != q]
for combo in combinations(others, antecedent_size):
candidates.append((frozenset(combo), q))
for A, q in candidates:
# --- Inference 1: Negative monotonicity (Qmax) ---
if q in minimal_global and q not in A:
N_current.add((A, q))
stats.inferred_negative_monotonicity += 1
log.append(
QueryAnswer(
Query.group(A, q),
False,
InferenceSource.MONOTONICITY_NEGATIVE,
)
)
continue
# --- Inference 2: Positive monotonicity ---
# If any proper subset of A (from prior blocks) implies q, then A implies q.
inferred_positive = False
for size in range(1, len(A)):
for sub_combo in combinations(A, size):
sub = frozenset(sub_combo)
if (sub, q) in P_current:
inferred_positive = True
break
if inferred_positive:
break
if inferred_positive:
stats.inferred_positive_monotonicity += 1
log.append(
QueryAnswer(
Query.group(A, q),
True,
InferenceSource.MONOTONICITY_POSITIVE,
)
)
if algorithm == "pending":
# An inferred positive is a positive answer like any other:
# its removal must be attempted, because its premise may
# itself be pending (then D(A, q) ⊆ D(premise) can be
# admissible even though the premise's removal is not).
_handle_positive(A, q)
continue
if algorithm == "one_pass":
# --- Structural test (cf. Theorem 16.1.6), one-pass only ---
# Simulate YES: remove states containing q but disjoint from A
DK = {state for state in L_current if q in state and not (A & state)}
L_potential = L_current - DK
if not _is_learning_space(L_potential, domain):
N_current.add((A, q))
stats.inferred_structural += 1
log.append(
QueryAnswer(
Query.group(A, q),
False,
InferenceSource.STRUCTURAL,
)
)
continue
# --- Expert query ---
query = Query.group(A, q)
answer = expert(query)
stats.expert_queries += 1
log.append(QueryAnswer(query, answer, InferenceSource.EXPERT))
if answer:
P_current.add((A, q))
if algorithm == "pending":
_handle_positive(A, q)
else:
# Admissibility was established by the structural test above.
L_current = {state for state in L_current if not (q in state and not (A & state))}
else:
N_current.add((A, q))
result_structure = KnowledgeStructure(items, L_current)
return BlockNResult(
structure=result_structure,
positive=P_current,
negative=N_current,
stats=stats,
log=log,
pending=tuple(pending),
)