"""Course-dependent skill and learning-object structures.
The CDSS workflow (Hockemeyer, CDSS 0.3-1, 2026) starts from taught/required
skill assignments. This independent implementation uses the attribution
semantics of Falmagne & Doignon (2011), Eq. 5.2, with explicit alternatives,
canonical surmise functions and preserved learning-object identifiers.
"""
from __future__ import annotations
from collections.abc import Collection, Mapping, Sequence
from dataclasses import dataclass
from knowledgespaces.structures.attribution import Attribution, _positive_limit
from knowledgespaces.structures.relations import SurmiseRelation
from knowledgespaces.structures.surmise_function import SurmiseFunction
def _labels(values: Collection[str], name: str) -> frozenset[str]:
if any(not isinstance(q, str) or not q for q in values):
raise ValueError(f"{name} must contain nonempty string labels.")
return frozenset(values)
[docs]
@dataclass(frozen=True)
class LearningObject:
"""One learning-object requirement alternative.
``taught`` skills are acquired together; ``required`` skills are jointly
required. Several records with the same identifier express alternative
requirements, but must agree on the taught skills. Inputs are copied to
frozensets; diagnostic violations are allowed until derivation is requested.
"""
identifier: str
taught: frozenset[str]
required: frozenset[str] = frozenset()
def __post_init__(self) -> None:
if not isinstance(self.identifier, str) or not self.identifier:
raise ValueError("A learning object requires a nonempty string identifier.")
object.__setattr__(self, "taught", _labels(self.taught, "taught"))
object.__setattr__(self, "required", _labels(self.required, "required"))
[docs]
@dataclass(frozen=True)
class AssignmentDiagnostics:
"""The three CDSS compliance conditions, with offending labels retained.
These are not a test of acyclicity. A cycle involving several objects
can pass all three conditions; a completion conflict can also arise from
co-teaching/redundancy without a directed prerequisite cycle.
"""
untaught_skills: frozenset[str]
nonteaching_objects: frozenset[str]
self_requirements: tuple[tuple[str, frozenset[str]], ...]
@property
def compliant(self) -> bool:
return not (self.untaught_skills or self.nonteaching_objects or self.self_requirements)
[docs]
@dataclass(frozen=True)
class CurriculumReachability:
"""One feasible object sequence and the maximal skills reachable from a seed.
Required skills are tested before each object's taught skills are added.
This is a deterministic feasibility computation, not a stochastic learning
model or an optimal curriculum. Objects in ``blocked_objects`` cannot be
scheduled from the given initial skills under these monotone requirements.
"""
sequence: tuple[str, ...]
mastered_skills: frozenset[str]
blocked_objects: frozenset[str]
[docs]
class CurriculumCompletionError(ValueError):
"""Strict prerequisite completion includes skills taught by the same object."""
[docs]
@dataclass(frozen=True)
class CurriculumResult:
"""Completed assignments and canonical structures on both labelled domains.
Relations are provided exactly when the respective function is
quasi-ordinal, including equivalent items. No state span is materialized.
``allow_cycles`` records whether joint-state completion was requested.
"""
original: SkillAssignment
completed: SkillAssignment
skill_function: SurmiseFunction
object_function: SurmiseFunction
skill_relation: SurmiseRelation | None
object_relation: SurmiseRelation | None
allow_cycles: bool
[docs]
class SkillAssignment:
"""Taught skills and alternative required skill sets for learning objects.
Each :class:`LearningObject` record is one alternative. Distinct records
for the same object must have identical taught skills. Duplicates are
removed, while alternative order and first-occurrence object order are
retained. A single assignment has one requirement set per object; a
multi-assignment has at least one object with alternatives.
The optional explicit ``skills`` domain preserves unused skills for
diagnostics. Syntax is checked at construction; semantic noncompliance
can be inspected with :meth:`diagnostics` before derivation, which rejects it.
"""
__slots__ = ("_objects", "_required", "_rows", "_skills", "_taught")
def __init__(
self, objects: Collection[LearningObject], *, skills: Collection[str] | None = None
) -> None:
rows = tuple(objects)
if not rows or any(not isinstance(row, LearningObject) for row in rows):
raise ValueError("An assignment requires LearningObject records.")
rows = tuple(dict.fromkeys(rows))
declared = frozenset().union(*(row.taught | row.required for row in rows))
self._skills = declared if skills is None else _labels(skills, "skills")
if not self._skills or not declared <= self._skills:
raise ValueError("A nonempty skill domain must contain every taught/required skill.")
taught: dict[str, frozenset[str]] = {}
required: dict[str, list[frozenset[str]]] = {}
for row in rows:
if row.identifier in taught and taught[row.identifier] != row.taught:
raise ValueError("Alternatives of the same object must teach the same skills.")
taught[row.identifier] = row.taught
required.setdefault(row.identifier, []).append(row.required)
self._rows = rows
self._objects = tuple(taught)
self._taught = taught
self._required = {q: tuple(family) for q, family in required.items()}
@property
def rows(self) -> tuple[LearningObject, ...]:
return self._rows
@property
def learning_objects(self) -> tuple[str, ...]:
return self._objects
@property
def skills(self) -> frozenset[str]:
return self._skills
@property
def is_single_assignment(self) -> bool:
return all(len(family) == 1 for family in self._required.values())
[docs]
def taught_by(self, learning_object: str) -> frozenset[str]:
if learning_object not in self._taught:
raise ValueError("Unknown learning object.")
return self._taught[learning_object]
[docs]
def requirements_for(self, learning_object: str) -> tuple[frozenset[str], ...]:
"""Alternative jointly required skill sets; the empty set means no prerequisites."""
if learning_object not in self._required:
raise ValueError("Unknown learning object.")
return self._required[learning_object]
[docs]
def diagnostics(self) -> AssignmentDiagnostics:
taught = frozenset().union(*self._taught.values())
overlaps = []
for q in self._objects:
overlap = self._taught[q] & frozenset().union(*self._required[q])
if overlap:
overlaps.append((q, overlap))
return AssignmentDiagnostics(
self._skills - taught,
frozenset(q for q in self._objects if not self._taught[q]),
tuple(overlaps),
)
def _require_compliance(self) -> None:
report = self.diagnostics()
if not report.compliant:
raise ValueError(f"Noncompliant skill assignment: {report}")
@property
def has_unique_teachers(self) -> bool:
"""Exactly one distinct teaching object per skill, including full coverage.
For a single assignment this implies quasi-ordinal derived structures.
For a multi-assignment, alternative requirements may still produce a
non-quasi-ordinal structure despite unique teaching objects.
"""
return all(sum(s in self._taught[q] for q in self._objects) == 1 for s in self._skills)
[docs]
def skill_attribution(self) -> Attribution:
"""Each taught skill receives its object's taught-plus-required sets as clauses."""
self._require_compliance()
return Attribution(
self._skills,
{
s: [row.taught | row.required for row in self._rows if s in row.taught]
for s in self._skills
},
)
[docs]
def object_attribution(self, *, max_candidates: int = 100_000) -> Attribution:
"""Alternative prerequisite object sets, with one consistent identifier per object.
For each required skill choose a teaching object; combine and minimize
their unions, then include the target object. This is the attribution
specified by ``CDSS::cdss_lo_sa2af``, extended to requirement alternatives.
"""
self._require_compliance()
_positive_limit(max_candidates, "max_candidates")
teachers = {
s: frozenset(frozenset({q}) for q in self._objects if s in self._taught[q])
for s in self._skills
}
clauses = {}
for q in self._objects:
family: set[frozenset[str]] = set()
for required in self._required[q]:
for cover in _covers(required, teachers, max_candidates):
family.add(cover | {q})
_bounded(family, max_candidates)
clauses[q] = _minimal(family)
return Attribution(self._objects, clauses)
[docs]
def is_complete(self, *, allow_cycles: bool = False) -> bool:
"""Whether every required skill has an included teaching/requirement alternative.
In strict completion the witness is included in the required skill
set alone. With ``allow_cycles=True`` it may also use skills taught by
the current object. This tests closure, independently of compliance.
"""
_cycle_flag(allow_cycles)
for row in self._rows:
available = row.required | row.taught if allow_cycles else row.required
for s in row.required:
if not any(
s in witness.taught and witness.taught | witness.required <= available
for witness in self._rows
):
return False
return True
[docs]
def complete(
self, *, allow_cycles: bool = False, max_candidates: int = 100_000
) -> SkillAssignment:
"""Complete all prerequisite alternatives, retaining inclusion-minimal results.
Strict completion requires complete prerequisite skill states disjoint
from the object's taught skills. An overlap raises
:class:`CurriculumCompletionError`; it can reflect cycles or redundant
co-teaching, not necessarily a directed graph cycle.
With ``allow_cycles=True``, complete taught-plus-required sets jointly,
then remove the object's taught skills from the recorded prerequisites.
This gives a defined static skill-space semantics for cycles; it is
not a claim that these objects can be scheduled from an empty skill set.
Use :meth:`reachability` for that separate question. Redundant original
requirement alternatives are minimized before the overlap check.
"""
self._require_compliance()
_cycle_flag(allow_cycles)
_positive_limit(max_candidates, "max_candidates")
function = self.skill_attribution().to_surmise_function(max_candidates=max_candidates)
providers = {s: function.clauses_for(s) for s in self._skills}
rows = []
for q in self._objects:
taught = self._taught[q]
family: set[frozenset[str]] = set()
for required in self._required[q]:
seed = required | taught if allow_cycles else required
for state in _covers(seed, providers, max_candidates):
family.add(state - taught if allow_cycles else state)
_bounded(family, max_candidates)
for required in sorted(_minimal(family), key=lambda s: (len(s), sorted(s))):
if required & taught:
raise CurriculumCompletionError(
f"Completion of {q!r} requires skills it also teaches: "
f"{sorted(required & taught)}. Inspect cycles/co-teaching or "
"request explicit joint-state completion with allow_cycles=True."
)
rows.append(LearningObject(q, taught, required))
return SkillAssignment(rows, skills=self._skills)
[docs]
def derive(
self, *, allow_cycles: bool = False, max_candidates: int = 100_000
) -> CurriculumResult:
"""Complete the course and derive canonical functions on skills and objects.
The object function closes the original labelled object attribution,
preserving alternatives without treating their duplicate rows as new
objects. No state enumeration or empirical validation is implied.
"""
completed = self.complete(allow_cycles=allow_cycles, max_candidates=max_candidates)
skill_function = completed.skill_attribution().to_surmise_function(
max_candidates=max_candidates
)
object_function = self.object_attribution(
max_candidates=max_candidates
).to_surmise_function(max_candidates=max_candidates)
return CurriculumResult(
self,
completed,
skill_function,
object_function,
skill_function.to_surmise_relation() if skill_function.is_quasi_ordinal else None,
object_function.to_surmise_relation() if object_function.is_quasi_ordinal else None,
allow_cycles,
)
[docs]
def reachability(self, initial_skills: Collection[str] = ()) -> CurriculumReachability:
"""Compute a feasible object order and all reachable skills by forward chaining.
Unknown initial skills are rejected. Noncompliant assignments can be
inspected here; for example an untaught prerequisite remains blocked
unless it is explicitly supplied among the initial skills.
"""
mastered = _labels(initial_skills, "initial_skills")
if not mastered <= self._skills:
raise ValueError("Initial skills are outside the declared domain.")
remaining = list(self._objects)
sequence = []
while True:
found = next(
(q for q in remaining if any(r <= mastered for r in self._required[q])), None
)
if found is None:
break
remaining.remove(found)
sequence.append(found)
mastered |= self._taught[found]
return CurriculumReachability(tuple(sequence), mastered, frozenset(remaining))
[docs]
@classmethod
def from_pairs(
cls,
taught: Collection[tuple[str, str]],
required: Collection[tuple[str, str]] = (),
*,
learning_objects: Collection[str] | None = None,
skills: Collection[str] | None = None,
) -> SkillAssignment:
"""Construct a single assignment from (object, skill) tables, as in CDSS.
Optional explicit domains retain empty objects/unused skills for
diagnostics. Repeated pairs are harmless. Labels are never stripped,
truncated or normalized; whitespace can be part of an identifier.
"""
pairs = [*taught, *required]
for pair in pairs:
if isinstance(pair, str) or len(pair) != 2:
raise ValueError("Each assignment pair must contain an object and a skill.")
_labels(pair, "assignment pairs")
inferred = tuple(dict.fromkeys(q for q, _ in pairs))
objects = inferred if learning_objects is None else tuple(learning_objects)
_labels(objects, "learning_objects")
if len(set(objects)) != len(objects) or not set(inferred) <= set(objects):
raise ValueError("Object domain must contain each table object exactly once.")
rows = [
LearningObject(
q,
frozenset(s for obj, s in taught if obj == q),
frozenset(s for obj, s in required if obj == q),
)
for q in objects
]
return cls(rows, skills=skills)
[docs]
@classmethod
def from_matrices(
cls,
learning_objects: Sequence[str],
skills: Sequence[str],
taught: Sequence[Sequence[int]],
required: Sequence[Sequence[int]],
) -> SkillAssignment:
"""Construct a single assignment from aligned binary object-by-skill matrices."""
_labels(skills, "skills")
_labels(learning_objects, "learning_objects")
if len(set(skills)) != len(skills) or len(set(learning_objects)) != len(learning_objects):
raise ValueError("Matrix labels must be unique.")
for matrix in (taught, required):
if len(matrix) != len(learning_objects) or any(
len(row) != len(skills) or any(v not in (0, 1) for v in row) for row in matrix
):
raise ValueError(
"Taught/required matrices must be binary and match the label dimensions."
)
rows = [
LearningObject(
q,
frozenset(s for s, bit in zip(skills, t, strict=True) if bit),
frozenset(s for s, bit in zip(skills, r, strict=True) if bit),
)
for q, t, r in zip(learning_objects, taught, required, strict=True)
]
return cls(rows, skills=skills)
[docs]
def to_matrices(self) -> tuple[list[str], list[str], list[list[int]], list[list[int]]]:
"""Return object labels, skill labels, taught matrix and required matrix.
This two-matrix layout requires one requirement set per object;
use ``rows`` to preserve alternatives of a multi-assignment.
"""
if not self.is_single_assignment:
raise ValueError(
"A multi-assignment needs alternative rows, not a single required matrix."
)
skills = sorted(self._skills)
return (
list(self._objects),
skills,
[[int(s in self._taught[q]) for s in skills] for q in self._objects],
[[int(s in self._required[q][0]) for s in skills] for q in self._objects],
)
def __repr__(self) -> str:
return (
f"SkillAssignment(n_objects={len(self._objects)}, n_skills={len(self._skills)}, "
f"n_alternatives={len(self._rows)})"
)
def _minimal(family: Collection[frozenset[str]]) -> frozenset[frozenset[str]]:
minimal: set[frozenset[str]] = set()
for state in sorted(family, key=lambda s: (len(s), sorted(s))):
if not any(s <= state for s in minimal):
minimal.add(state)
return frozenset(minimal)
def _bounded(family: Collection[frozenset[str]], limit: int) -> None:
if len(family) > limit:
raise ValueError("Curriculum alternatives exceed max_candidates.")
def _covers(
required: frozenset[str], providers: Mapping[str, frozenset[frozenset[str]]], limit: int
) -> frozenset[frozenset[str]]:
family: frozenset[frozenset[str]] = frozenset({frozenset()})
for skill in sorted(required):
expanded: set[frozenset[str]] = set()
for a in family:
for b in providers[skill]:
expanded.add(a | b)
_bounded(expanded, limit)
family = _minimal(expanded)
return family
def _cycle_flag(value: bool) -> None:
if not isinstance(value, bool):
raise ValueError("allow_cycles must be a bool.")