"""Explicit, lossless conversions of binary patterns and named state rows.
These in-memory operations preserve row order and duplicates. Use SetFamily
to request deduplication, and KnowledgeStructure to insert endpoint states.
"""
from __future__ import annotations
from collections.abc import Collection, Sequence
from typing import Literal
import numpy as np
from numpy.typing import DTypeLike
from knowledgespaces._patterns import _positive_integer
from knowledgespaces.estimation.blim_em import ResponseMatrix
from knowledgespaces.structures.knowledge_base import KnowledgeBase
from knowledgespaces.structures.knowledge_structure import KnowledgeStructure
def _labels(items: Sequence[str]) -> tuple[str, ...]:
labels = tuple(items)
if isinstance(items, str) or any(not isinstance(q, str) or not q for q in labels):
raise ValueError("items must be a sequence of nonempty string labels.")
if len(labels) != len(set(labels)):
raise ValueError("items must be unique.")
return labels
def _binary(matrix: np.ndarray) -> np.ndarray:
rows = np.asarray(matrix)
if rows.ndim != 2 or rows.dtype.kind not in "biuf" or not np.isin(rows, [0, 1]).all():
raise ValueError("Expected a two-dimensional binary numeric matrix.")
return rows
def _dtype(dtype: DTypeLike) -> np.dtype:
result = np.dtype(dtype)
if result.kind not in "biuf":
raise ValueError("dtype must be boolean, integer or floating point.")
return result
def _size(nrows: int, ncols: int, max_cells: int) -> None:
if max(nrows, nrows * ncols) > _positive_integer(max_cells, "max_cells"):
raise ValueError("Output exceeds max_cells; no truncated result is returned.")
[docs]
def states_to_matrix(
states: Collection[Collection[str]],
*,
items: Sequence[str],
dtype: DTypeLike = "int8",
max_cells: int = 1_000_000,
) -> np.ndarray:
"""Encode state rows in the supplied column order, without adding endpoints.
A path/list retains its order and duplicates; SetFamily and
KnowledgeStructure iterate in their canonical order. Empty rows/columns
are supported, including shape (0, len(items)). Unknown items raise.
``dtype=bool`` provides a logical matrix. No labels are inferred from
observed states, so never-observed items remain represented.
"""
labels, output_type = _labels(items), _dtype(dtype)
_size(len(states), len(labels), max_cells)
domain = frozenset(labels)
result = np.zeros((len(states), len(labels)), dtype=output_type)
for i, state in enumerate(states):
if isinstance(state, str) or not frozenset(state) <= domain:
raise ValueError("Each state must be a collection of items from the domain.")
result[i] = [q in state for q in labels]
return result
[docs]
def matrix_to_states(matrix: np.ndarray, *, items: Sequence[str]) -> tuple[frozenset[str], ...]:
"""Decode rows preserving order and duplicates, with explicit column labels."""
labels, rows = _labels(items), _binary(matrix)
if rows.shape[1] != len(labels):
raise ValueError("items must label every matrix column.")
return tuple(frozenset(q for q, bit in zip(labels, row, strict=True) if bit) for row in rows)
def _text_options(labels: tuple[str, ...], separator: str, empty: str) -> None:
if not isinstance(separator, str) or not isinstance(empty, str):
raise ValueError("separator and empty must be strings.")
if (separator and any(separator in q for q in labels)) or (
not separator and any(len(q) != 1 for q in labels)
):
raise ValueError(
"Use a separator absent from every label; no separator needs one-character labels."
)
tokens = empty.split(separator) if separator else list(empty)
if empty and all(token in labels for token in tokens):
raise ValueError("empty must not be confused with a nonempty labelled pattern.")
[docs]
def matrix_to_patterns(
matrix: np.ndarray,
*,
items: Sequence[str] | None = None,
separator: str = "",
empty: str = "{}",
) -> tuple[str, ...]:
"""Encode rows as binary strings, or labelled strings when items is supplied.
The latter joins correct item labels with separator and uses empty for
the empty response. Ambiguous label/separator/empty combinations raise
rather than silently losing information. Duplicates and order survive.
Binary strings use exactly one 0/1 per column, independent of separator.
"""
rows = _binary(matrix)
if items is None:
return tuple("".join("1" if bit else "0" for bit in row) for row in rows)
labels = _labels(items)
_text_options(labels, separator, empty)
return tuple(
separator.join(q for q in labels if q in state) or empty
for state in matrix_to_states(rows, items=labels)
)
[docs]
def patterns_to_matrix(
patterns: Sequence[str],
*,
items: Sequence[str] | None = None,
separator: str = "",
empty: str = "{}",
dtype: DTypeLike = "int8",
max_cells: int = 1_000_000,
) -> np.ndarray:
"""Decode binary strings, or labelled patterns with an explicit domain.
With items=None, infer the binary width from the first row; an empty
sequence has shape (0,0). With items supplied, parse labelled strings
using the same conventions as matrix_to_patterns. Repeated labels,
unknown labels and ragged/nonbinary rows are rejected.
"""
if isinstance(patterns, str) or any(not isinstance(p, str) for p in patterns):
raise ValueError("patterns must be a sequence of strings.")
output_type = _dtype(dtype)
if items is None:
width = len(patterns[0]) if len(patterns) else 0
_size(len(patterns), width, max_cells)
if any(len(p) != width or any(bit not in "01" for bit in p) for p in patterns):
raise ValueError("Binary patterns must have equal width and contain only 0/1.")
return np.array([[int(bit) for bit in p] for p in patterns], dtype=output_type).reshape(
len(patterns), width
)
labels = _labels(items)
_text_options(labels, separator, empty)
_size(len(patterns), len(labels), max_cells)
states = []
for pattern in patterns:
tokens = (
[] if pattern == empty else (pattern.split(separator) if separator else list(pattern))
)
if len(tokens) != len(set(tokens)) or not set(tokens) <= set(labels):
raise ValueError("Labelled patterns must contain distinct known items.")
states.append(frozenset(tokens))
return states_to_matrix(states, items=labels, dtype=output_type, max_cells=max_cells)
[docs]
def expand_response_patterns(
data: ResponseMatrix,
*,
max_rows: int = 1_000_000,
max_cells: int = 1_000_000,
dtype: DTypeLike = "int8",
) -> np.ndarray:
"""Expand integer response frequencies, retaining original row/column order.
Fractional weights cannot represent individual respondents and raise.
Output rows/cells are checked before repetition; zero-count rows vanish.
For analysis without expansion keep the weighted ResponseMatrix instead.
"""
rows = _binary(data.patterns)
_labels(data.items)
checked = ResponseMatrix(list(data.items), rows, data.counts)
weights = checked.effective_counts
if not np.equal(weights, np.floor(weights)).all():
raise ValueError("Expansion requires integer counts, not fractional weights.")
total = sum(int(w) for w in weights)
if total > _positive_integer(max_rows, "max_rows"):
raise ValueError("Expansion exceeds max_rows.")
_size(total, rows.shape[1], max_cells)
return np.repeat(rows.astype(_dtype(dtype)), weights.astype(np.intp), axis=0)
[docs]
def subset_matrix(
states: Collection[Collection[str]],
other: Collection[Collection[str]] | None = None,
*,
proper: bool = False,
max_cells: int = 1_000_000,
) -> np.ndarray:
"""Boolean incidence I[i,j] = states[i] <= other[j] (pks is.subset).
other=None compares states with themselves; proper=True uses strict
inclusion. Rows/columns follow input iteration, including duplicates.
This is inclusion incidence, not the adjacency of a Hasse diagram.
"""
if not isinstance(proper, bool):
raise ValueError("proper must be a bool.")
left = tuple(frozenset(s) for s in states)
right = left if other is None else tuple(frozenset(s) for s in other)
_size(len(left), len(right), max_cells)
return np.array(
[[a < b if proper else a <= b for b in right] for a in left], dtype=bool
).reshape(len(left), len(right))
[docs]
def boolean_matrix_product(
left: np.ndarray, right: np.ndarray, *, max_cells: int = 1_000_000
) -> np.ndarray:
"""Boolean relational composition: any intermediate link, without integer overflow."""
a, b = _binary(left), _binary(right)
if a.shape[1] != b.shape[0]:
raise ValueError("Inner matrix dimensions must agree.")
_size(a.shape[0], b.shape[1], max_cells)
return a.astype(bool) @ b.astype(bool)
[docs]
def fringe_matrix(
structure: KnowledgeStructure | KnowledgeBase,
states: Collection[Collection[str]],
*,
items: Sequence[str] | None = None,
kind: Literal["inner", "outer", "both"] = "both",
max_cells: int = 1_000_000,
) -> np.ndarray:
"""Binary fringe rows for explicit states, using a structure or compact base.
Both means the union of inner and outer fringes. Each input must be a
state of the represented structure; no span is enumerated for a base.
Columns default to sorted domain and rows preserve input order.
"""
labels = _labels(sorted(structure.domain) if items is None else items)
if set(labels) != structure.domain or kind not in ("inner", "outer", "both"):
raise ValueError("items must order the whole domain and kind must be inner/outer/both.")
_size(len(states), len(labels), max_cells)
result = []
for state in states:
row = frozenset(state)
if row not in structure:
raise ValueError("Every row must be a state of the represented structure.")
fringe: frozenset[str] = frozenset()
if kind != "outer":
fringe = structure.inner_fringe(row)
if kind != "inner":
fringe |= structure.outer_fringe(row)
result.append(fringe)
return states_to_matrix(result, items=labels, max_cells=max_cells)