"""Observed response-pattern frequencies, independent of latent-state fitting."""
from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass
import numpy as np
from knowledgespaces._patterns import _positive_integer
from knowledgespaces.estimation.blim_em import ResponseMatrix
from knowledgespaces.structures.knowledge_structure import KnowledgeStructure
[docs]
@dataclass(frozen=True)
class PatternFrequencies:
"""Frequency report with read-only arrays and explicit item order.
patterns/counts contain the selected most frequent observed rows.
probabilities divides these counts by total_weight, including rows
omitted by n. n_observed_patterns counts all distinct positive-weight
rows. reference_patterns/reference_counts retain the requested row
order and duplicates, with zero for rows absent from the observations.
Matching a knowledge state here is an observed response frequency,
not an estimate of that state's latent probability under a BLIM.
"""
items: tuple[str, ...]
patterns: np.ndarray
counts: np.ndarray
probabilities: np.ndarray
total_weight: float
n_observed_patterns: int
reference_patterns: np.ndarray | None
reference_counts: np.ndarray | None
@property
def n(self) -> int:
"""Number of returned observed patterns, after truncation."""
return len(self.counts)
def _rows(value: np.ndarray, name: str) -> np.ndarray:
rows = np.asarray(value)
if rows.ndim != 2 or rows.dtype.kind not in "biuf" or not np.isfinite(rows).all():
raise ValueError(f"{name} must be a finite, real numeric two-dimensional matrix.")
return rows
[docs]
def pattern_frequencies(
data: ResponseMatrix | np.ndarray,
*,
n: int | None = 5,
reference_patterns: np.ndarray | KnowledgeStructure | None = None,
items: Sequence[str] | None = None,
counts: np.ndarray | None = None,
max_memory_bytes: int = 512_000_000,
) -> PatternFrequencies:
"""Count observed patterns and exact matches to optional reference rows.
This is the descriptive workflow of DAKS::pattern(dataset,n,P), with
weights and unambiguous numeric row keys. ResponseMatrix supplies its
labels and frequencies; items/counts must then be omitted. Raw numeric
matrices also support polytomous scores (this does not extend BLIM or
IITA to such scores). Raw labels default to item_1, item_2, etc.
n=None returns every distinct positive-weight row. Otherwise n is a
positive integer, capped at that number. Counts sort descending, ties
by ascending numeric row order. Fractions use the total input weight,
not the truncated sum. Nonnegative fractional weights are allowed;
zero-weight rows are excluded from the observed-pattern report.
reference_patterns is a matrix in data's column order, or a labelled
KnowledgeStructure (aligned by label, in canonical state order).
Every reference row is retained, including duplicates and absent rows.
Missing/infinite scores are rejected rather than imputed or omitted.
Inputs are copied into read-only output arrays; integral scores are
not converted to float, and patterns are never concatenated into strings.
max_memory_bytes preflights estimated array and row-lookup storage;
it is not an exact peak-memory guarantee of NumPy's sorting backend.
"""
limit = None if n is None else _positive_integer(n, "n")
max_memory_bytes = _positive_integer(max_memory_bytes, "max_memory_bytes")
if isinstance(data, ResponseMatrix):
if items is not None or counts is not None:
raise ValueError("ResponseMatrix supplies items and counts; omit separate arguments.")
# Revalidate potentially mutated public arrays without changing them.
data = ResponseMatrix(list(data.items), data.patterns, data.counts)
labels = list(data.items)
rows = _rows(data.patterns, "data")
weights = data.effective_counts
else:
rows = _rows(data, "data")
labels = [f"item_{i + 1}" for i in range(rows.shape[1])] if items is None else list(items)
weights = np.ones(len(rows)) if counts is None else np.asarray(counts)
if (
not labels
or len(labels) != rows.shape[1]
or any(not isinstance(q, str) for q in labels)
or len(labels) != len(set(labels))
):
raise ValueError("items must contain one distinct string label per nonempty column.")
if (
weights.shape != (len(rows),)
or weights.dtype.kind not in "biuf"
or not np.isfinite(weights).all()
or np.any(weights < 0)
):
raise ValueError("counts must be a finite nonnegative vector matching data rows.")
weights = weights.astype(float)
with np.errstate(over="ignore"):
total = float(weights.sum())
if not np.isfinite(total) or total <= 0:
raise ValueError("Total count must be finite and positive.")
base_bytes = 5 * rows.nbytes + 64 * rows.size + 40 * len(rows)
reference = None
if isinstance(reference_patterns, KnowledgeStructure):
if reference_patterns.domain != set(labels):
raise ValueError("Reference structure domain must match data items.")
if base_bytes + 65 * len(reference_patterns) * len(labels) > max_memory_bytes:
raise MemoryError("Pattern frequency working storage exceeds max_memory_bytes.")
reference = np.array(
[[q in state for q in labels] for state in reference_patterns], dtype=np.int8
)
elif reference_patterns is not None:
reference = _rows(reference_patterns, "reference_patterns")
if reference.shape[1] != len(labels):
raise ValueError("reference_patterns must have the same columns as data.")
extra = 0 if reference is None else reference.nbytes + 64 * reference.size
if base_bytes + extra > max_memory_bytes:
raise MemoryError("Pattern frequency working storage exceeds max_memory_bytes.")
unique, inverse = np.unique(rows[weights > 0], axis=0, return_inverse=True)
frequencies = np.bincount(inverse.ravel(), weights=weights[weights > 0], minlength=len(unique))
selected = np.argsort(-frequencies, kind="stable")[:limit]
reference_counts = None
if reference is not None:
lookup = {
tuple(row.tolist()): float(weight)
for row, weight in zip(unique, frequencies, strict=True)
}
reference_counts = np.array([lookup.get(tuple(row.tolist()), 0.0) for row in reference])
reference = reference.copy()
reference.flags.writeable = reference_counts.flags.writeable = False
patterns, selected_counts = unique[selected].copy(), frequencies[selected].copy()
probabilities = selected_counts / total
for array in (patterns, selected_counts, probabilities):
array.flags.writeable = False
return PatternFrequencies(
tuple(labels),
patterns,
selected_counts,
probabilities,
total,
len(unique),
reference,
reference_counts,
)