"""Assessment selection and recording without unintentional question reuse."""
import csv
import random
from collections import Counter
from collections.abc import Sequence, Set
from pathlib import Path
from .assessments import assessment_name, is_eligible
from .models import Bank, PlannedUse, Question, UsageRecord
HISTORY_FIELDS = ["course", "bank_id", "term", "assessment", "id"]
[docs]
def read_history(path: str | Path, bank: Bank) -> list[UsageRecord]:
"""Read recorded question use for the supplied course and bank.
Parameters
----------
path
Usage-history CSV encoded as UTF-8, with an optional byte-order mark.
A missing file or a zero-byte file is treated as empty history.
bank
Validated bank whose course and bank ID identify the rows to return.
Other banks' rows are skipped.
Returns
-------
records
New usage records in file order. A legacy ``exam`` column is exposed
as ``assessment``. Additional CSV columns are omitted from records.
Raises
------
ValueError
If the header is invalid, a matching row lacks a value for a declared
column, or a matching row has a blank assessment name.
OSError
If the file cannot be inspected or read.
UnicodeError
If the file cannot be decoded as UTF-8.
csv.Error
If CSV parsing fails.
Notes
-----
Reading does not modify the file or bank. It does not check whether
recorded question IDs still exist in the bank or reject repeated IDs.
Column presence is checked for matching rows; empty cell text is allowed
except for the assessment name.
""" # noqa: DOC503
path = Path(path)
if not path.exists() or path.stat().st_size == 0:
return []
with path.open(newline="", encoding="utf-8-sig") as stream:
reader = csv.DictReader(stream)
fields = _history_fields(reader.fieldnames)
assessment_field = "assessment" if "assessment" in fields else "exam"
records: list[UsageRecord] = []
for row in reader:
if (
row["course"] != bank["course"]
or row["bank_id"] != bank["bank_id"]
):
continue
if any(not isinstance(row.get(field), str) for field in fields):
raise ValueError("Usage history contains an incomplete row")
records.append(
UsageRecord(
course=row["course"],
bank_id=row["bank_id"],
term=row["term"],
assessment=assessment_name(row[assessment_field]),
id=row["id"],
)
)
return records
[docs]
def choose(
bank: Bank, assessment: str, count: int, used: Set[str], seed: str | int = 0
) -> list[Question]:
"""Propose unused questions, balancing areas before difficulties.
Parameters
----------
bank
Validated bank containing the candidate questions.
assessment
Nonblank assessment name, matched exactly, including case and
whitespace. Questions with unrestricted eligibility may also be used.
count
Nonnegative integer number of questions to select. Zero returns an
empty list; booleans and non-integer values are rejected.
used
Question IDs to exclude. The caller supplies these IDs, typically
from this bank's usage history; no history file is read here.
seed
Seed for local random tie breaking, converted to text. Defaults to 0.
Results are reproducible for the same inputs and question order.
Returns
-------
questions
Selected questions in selection order, without repetition. Entries
reference the original bank dictionaries rather than copies.
Raises
------
ValueError
If the assessment name or count is invalid, or fewer eligible,
unused questions remain than requested.
Notes
-----
Selection is greedy. At each step, prefer the least represented area in
the current selection, then the least represented difficulty, then break
ties randomly. This is a balancing heuristic, not a coverage guarantee.
The bank and used set are unchanged, and no usage is recorded.
"""
assessment_name(assessment)
_validate_count(count)
candidates = [
q
for q in bank["questions"]
if q["id"] not in used and is_eligible(q, assessment)
]
if len(candidates) < count:
raise ValueError(
f"Only {len(candidates)} unused {assessment}-eligible "
"questions remain; "
f"cannot choose {count} without reuse"
)
rand = random.Random(str(seed)) # ruff: ignore[suspicious-non-cryptographic-random-usage]
chosen: list[Question] = []
areas: Counter[str] = Counter()
difficulties: Counter[str] = Counter()
while len(chosen) < count:
scored: list[tuple[tuple[int, int, float], Question]] = []
for q in candidates:
if q in chosen:
continue
# Balance areas first, then difficulty; draw among equivalent
# candidates reproducibly.
score = (
areas[q["area"]],
difficulties[q["difficulty"]],
rand.random(),
)
scored.append((score, q))
q = min(scored, key=lambda p: p[0])[1]
chosen.append(q)
areas[q["area"]] += 1
difficulties[q["difficulty"]] += 1
return chosen
[docs]
def append_record(
path: str | Path, bank: Bank, term: str, assessment: str, ids: Sequence[str]
) -> None:
"""Append actually administered question IDs to usage history.
Parameters
----------
path
Destination CSV. Parent directories are created as needed. New or
zero-byte files receive the standard header; existing files retain
their column order and any legacy ``exam`` column.
bank
Validated bank supplying the course, bank ID, and eligible questions.
term
Nonblank label for when the questions were administered. Stored
exactly as supplied, without trimming.
assessment
Nonblank assessment name, matched exactly against question
eligibility and stored without normalization.
ids
Unique question IDs to append in the supplied order. Every ID must
exist in the bank, be eligible, and have no recorded use in this
course and bank, regardless of term or assessment. An empty sequence
is allowed and may still create directories and a header-only file.
Returns
-------
None
Return after appending all requested usage records.
Raises
------
ValueError
If the term or assessment is invalid, IDs are duplicate, unknown,
ineligible, or previously used, or existing history fails validation.
OSError
If reading history, creating directories, or writing records fails.
UnicodeError
If existing history cannot be decoded or output cannot be encoded.
csv.Error
If reading or writing CSV fails.
Notes
-----
Input and existing-history validation occur before appending records.
The bank is unchanged. Additional columns in existing history receive
empty values for new rows.
""" # noqa: DOC503
if not term.strip():
raise ValueError("term cannot be blank")
assessment_name(assessment)
valid = {q["id"] for q in bank["questions"] if is_eligible(q, assessment)}
if len(ids) != len(set(ids)) or any(q not in valid for q in ids):
raise ValueError(
"Question IDs must be unique and eligible for the specified "
"assessment"
)
existing = read_history(path, bank)
existing_ids = {x["id"] for x in existing}
if existing_ids.intersection(ids):
raise ValueError(
"Reusing previously recorded IDs is blocked: "
f"{sorted(existing_ids.intersection(ids))}"
)
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
fresh = not path.exists() or path.stat().st_size == 0
fields = HISTORY_FIELDS
if not fresh:
with path.open(newline="", encoding="utf-8-sig") as stream:
fields = _history_fields(csv.DictReader(stream).fieldnames)
assessment_field = "assessment" if "assessment" in fields else "exam"
with path.open("a", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fields)
if fresh:
writer.writeheader()
for q in ids:
writer.writerow({
"course": bank["course"],
"bank_id": bank["bank_id"],
"term": term,
assessment_field: assessment,
"id": q,
})
[docs]
def multi_term_plan(
bank: Bank,
terms: Sequence[str],
assessments: Sequence[tuple[str, int]],
used: Set[str] | None = None,
seed: str | int = 0,
) -> list[PlannedUse]:
"""Propose nonrepeating selections for named assessments in each term.
Parameters
----------
bank
Validated bank containing the candidate questions.
terms
Nonempty sequence of nonblank term labels, processed in order.
Labels are preserved exactly; repeated labels are permitted.
assessments
Nonempty sequence of ``(name, count)`` pairs. Names must be nonblank
and unique; counts must be nonnegative integers, excluding booleans.
Every pair applies to every term, in the supplied order.
used
Previously used question IDs to exclude throughout the plan.
Defaults to None, meaning no prior use. The supplied set is copied.
seed
Base seed for reproducible selection, defaulting to 0. Each call to
choose receives the text ``seed/term/assessment`` as its seed.
Returns
-------
plan
New records ordered by term, assessment, then question selection.
Each record contains course, bank ID, term, assessment, question ID,
area, and difficulty. A question ID appears at most once in the plan.
All-zero counts produce an empty list.
Raises
------
ValueError
If terms or assessments are empty, any label or count is invalid,
assessment names repeat, or a selection has insufficient candidates.
Notes
-----
Selections use choose's greedy balancing and exclude all earlier
selections in this proposal. Earlier choices are not reconsidered, so
failure does not prove that no alternative allocation is possible.
No history file is read or written, and caller-owned data is unchanged.
A failed plan raises an exception rather than returning a partial plan.
"""
names: set[str] = set()
for name, count in assessments:
assessment_name(name)
_validate_count(count)
if name in names:
raise ValueError("Plan assessment names must be unique")
names.add(name)
if not names:
raise ValueError("Plan needs at least one assessment")
if not terms:
raise ValueError("Plan needs at least one term")
if any(not term.strip() for term in terms):
raise ValueError("term cannot be blank")
selected_ids = set(used or [])
result: list[PlannedUse] = []
for term in terms:
for assessment, count in assessments:
chosen = choose(
bank,
assessment,
count,
selected_ids,
seed=f"{seed}/{term}/{assessment}",
)
for q in chosen:
selected_ids.add(q["id"])
result.append({
"course": bank["course"],
"bank_id": bank["bank_id"],
"term": term,
"assessment": assessment,
"id": q["id"],
"area": q["area"],
"difficulty": q["difficulty"],
})
return result
def _validate_count(count: object) -> None:
"""Reject negative counts and non-integers before selecting questions.
Parameters
----------
count
Requested number of questions. Zero is allowed. Booleans, strings,
and floating-point values are rejected without coercion.
Returns
-------
None
Return normally when the count is a nonnegative integer.
Raises
------
ValueError
If the count is negative, a boolean, or not an integer.
"""
if isinstance(count, bool) or not isinstance(count, int) or count < 0:
raise ValueError("count must be a nonnegative integer")
def _history_fields(fields: Sequence[str] | None) -> list[str]:
"""Validate new or legacy history headers without losing column order.
Parameters
----------
fields
CSV column names, or None when no header is available. Names must
include ``course``, ``bank_id``, ``term``, and ``id``, plus exactly
one of ``assessment`` or the legacy ``exam`` column. Additional
columns are allowed, but all names must be unique.
Returns
-------
names
A new list containing all supplied column names in their original
order. The input sequence is not modified.
Raises
------
ValueError
If the header is missing or empty, a required column is absent,
both assessment columns are present, or any column name is repeated.
Notes
-----
Column names are case-sensitive and are not stripped or normalized.
The legacy ``exam`` name is retained so appending records preserves
the existing CSV header.
"""
names = list(fields or [])
required = {"course", "bank_id", "term", "id"}
has_assessment = "assessment" in names
has_exam = "exam" in names
if (
not required.issubset(names)
or has_assessment == has_exam
or len(names) != len(set(names))
):
raise ValueError(f"Usage history needs columns {HISTORY_FIELDS}")
return names