"""
Full QUERY pipeline: Block 1 → L1 → Block 2 → ... → Block N.
Orchestrates the complete derivation of a learning space from expert queries.
"""
from __future__ import annotations
import warnings
from dataclasses import dataclass, field
from knowledgespaces._limits import DEFAULT_MAX_N_ITEMS, check_domain_size
from knowledgespaces.query.block1 import Block1Result, run_block1
from knowledgespaces.query.block2 import (
Algorithm,
BlockNResult,
PendingQuery,
run_block_n,
)
from knowledgespaces.query.expert import Expert
from knowledgespaces.structures.knowledge_structure import KnowledgeStructure
[docs]
class TruncatedQueryWarning(UserWarning):
"""The elicitation stopped before antecedents of size ``|Q| - 1``.
Latent learning spaces whose refinement requires larger antecedents
are returned as their depth-``max_antecedent_size`` approximation
(a superset of the latent space); the recovery guarantee of
Falmagne & Doignon (2011, Theorem 16.1.16) needs queries of every
antecedent size. Silence with ``warnings.filterwarnings`` when the
truncation depth is a deliberate part of the elicitation design.
"""
[docs]
@dataclass
class QueryPipelineResult:
"""Result of the full QUERY pipeline.
``pending_unapplied`` lists the positive answers whose state removal
was still inadmissible when the procedure ended (``"pending"``
algorithm only): the returned structure is then a learning space
consistent with all *applied* answers, but not with these. With an
error-free expert whose latent structure is a learning space this
can only happen under depth truncation.
"""
structure: KnowledgeStructure
block1: Block1Result
block_n_results: list[BlockNResult] = field(default_factory=list)
algorithm: str = "pending"
truncated: bool = False
pending_unapplied: tuple[PendingQuery, ...] = ()
@property
def total_expert_queries(self) -> int:
total = self.block1.stats.total_queries
for bn in self.block_n_results:
total += bn.stats.expert_queries
return total
@property
def is_learning_space(self) -> bool:
return self.structure.is_learning_space
@property
def recovery_guarantee_applies(self) -> bool:
"""Whether the algorithmic conditions of Theorem 16.1.16 hold.
True when the pending-table algorithm ran at full depth
(antecedents up to ``|Q| - 1``) and no positive answer was left
unapplied. Under these conditions, *assuming an error-free
expert whose latent structure is a learning space*, the returned
structure equals the latent space (Falmagne & Doignon, 2011,
Theorem 16.1.16). The expert-side assumptions cannot be checked
by the package.
"""
return self.algorithm == "pending" and not self.truncated and not self.pending_unapplied
[docs]
def run_query(
items: list[str],
expert: Expert,
*,
use_qmax: bool = True,
max_antecedent_size: int = 2,
max_items: int | None = None,
algorithm: Algorithm = "pending",
) -> QueryPipelineResult:
"""Run the full QUERY algorithm.
Parameters
----------
items : list[str]
The domain of items.
expert : Expert
The expert to query.
use_qmax : bool
If True, use Qmax minimality test in Block 1.
max_antecedent_size : int
Maximum antecedent size for group queries. Default 2 (Block 2 only).
Set to ``len(items) - 1`` for the full procedure; smaller values
truncate the elicitation and emit a :class:`TruncatedQueryWarning`.
max_items : int | None
Hard limit override for the ``|Q|`` preflight check. When None
(default), ``DEFAULT_MAX_N_ITEMS`` is used. QUERY is exponential
in ``|Q|``; values above ~25 are not practical in a browser worker.
algorithm : {"pending", "one_pass"}
Treatment of positive answers whose state removal is momentarily
inadmissible in Blocks 2+. The default ``"pending"`` buffers and
systematically re-tests them (Falmagne & Doignon 2011,
Section 16.1; recovery guarantee of Theorem 16.1.16 at full
depth); ``"one_pass"`` is the naive scheme of Algorithm 16.1.10,
which records them as structural negatives without consulting
the expert and can suppress truthful answers permanently.
Returns
-------
QueryPipelineResult
The derived learning space with full traceability.
Raises
------
ValueError
If ``items`` is empty. QUERY is defined only for a nonempty
domain (Koppen 1993, §2; Falmagne & Doignon 2011, §15).
Warns
-----
TruncatedQueryWarning
When ``max_antecedent_size < len(items) - 1``.
"""
if len(items) == 0:
raise ValueError(
"run_query requires a non-empty domain. "
"QUERY is defined on a nonempty set of items "
"(Koppen 1993, §2)."
)
check_domain_size(
len(items),
max_threshold=max_items if max_items is not None else DEFAULT_MAX_N_ITEMS,
context="run_query",
)
if algorithm not in ("pending", "one_pass"):
raise ValueError(f"algorithm must be 'pending' or 'one_pass', got {algorithm!r}")
full_depth = len(items) - 1
truncated = max_antecedent_size < full_depth
if truncated:
warnings.warn(
TruncatedQueryWarning(
f"run_query stops at antecedent size {max_antecedent_size} < "
f"|Q| - 1 = {full_depth}: the result is the depth-"
f"{max_antecedent_size} approximation of the latent structure "
f"(a superset when deeper refinement would apply). Pass "
f"max_antecedent_size={full_depth} for the full procedure."
),
stacklevel=2,
)
# Block 1: discover surmise relation
b1 = run_block1(items, expert, use_qmax=use_qmax, max_items=max_items)
# Generate ordinal space L1 from the closure
L1 = KnowledgeStructure.from_surmise_relation(b1.closure, max_items=max_items)
# Build initial positive set from Block 1 closure
prior_positive: set[tuple[frozenset[str], str]] = {(frozenset({a}), b) for a, b in b1.closure}
# Block 2, 3, ..., N: refine to learning space
current = L1
block_n_results: list[BlockNResult] = []
pending: tuple[PendingQuery, ...] = ()
for size in range(2, max_antecedent_size + 1):
bn = run_block_n(
items,
current,
prior_positive,
expert,
antecedent_size=size,
minimal_global=b1.minimal_global,
max_items=max_items,
algorithm=algorithm,
pending_in=pending,
)
block_n_results.append(bn)
current = bn.structure
prior_positive = bn.positive
pending = bn.pending
return QueryPipelineResult(
structure=current,
block1=b1,
block_n_results=block_n_results,
algorithm=algorithm,
truncated=truncated,
pending_unapplied=pending,
)