diff --git a/CHANGELOG.md b/CHANGELOG.md index 8a0e9b1..69c11e2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - Schema-validated score template registry with bundled `performance_health_v1` definition and FastAPI endpoints. - Automated score template tests covering registry usage and HTTP access. +- Adaptive blueprint engine with stratified selection, knockout-aware scoring, preview endpoints, and bundled sample pool. +- Assessment version catalog with item bank, response linkage, and exposure stats models plus seed script. ### Changed - Centralized SPOT documentation to include bundled templates and score templates. @@ -22,6 +24,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Docs - Added score template function documentation, ERD assets, and schema overview with nb-NO localization. +- Added blueprint engine function documentation, SPOT updates, ERD refresh, and localization alignment. +- Documented item bank models, refreshed schema diagrams, and updated SPOT/tasklist references. ## [3.1.0] - 2025-09-24 diff --git a/SPOT.md b/SPOT.md index 662530c..b076ef1 100644 --- a/SPOT.md +++ b/SPOT.md @@ -10,10 +10,23 @@ - [x] Expose score template summaries and definitions through FastAPI endpoints and automated tests. - [x] Update documentation set (function docs, tasklist, ERD, localization) and changelog for the new score template feature. +## Integration Plan – Adaptive Blueprint Selection +- [x] Add blueprint schema resources with anchors, quotas, and scoring metadata. +- [x] Implement stratified selector + scoring engine with exposure handling tests. +- [x] Publish blueprint preview API endpoints and automated coverage. +- [x] Refresh documentation (function doc, SPOT, ERD, localization) and changelog entries. + +## Integration Plan – Assessment Item Bank Tracking +- [x] Model persistent assessment versions, item bank entries, response linkage, and exposure statistics. +- [x] Seed Postgres schema for the new tables with constraints and indexes. +- [x] Extend blueprint engine utilities to hydrate pools from item bank records and stats. +- [x] Update documentation set (function doc, SPOT, ERD, changelog, localization, tasklist). + ## Function Catalog ### Backend (Python/FastAPI) - **Assessment Template Registry** — Bundled template loader & API. _(Function doc pending migration; tracked in integration plan.)_ +- **Assessment Blueprint Engine** — Stratified item blueprint loader, selector, scoring, and preview API. → `./docs/functions/assessment_blueprint_engine.md` - **Score Template Registry** — Schema-validated score template loader with FastAPI exposure. → `./docs/functions/score_template_registry.md` - **API Router v1** — Aggregates versioned public/admin routers. → `./docs/functions/api_router_v1.md` - **Core Setup & Lifespan** — FastAPI factory & Redis pool orchestration with test bypass flag. → `./docs/functions/core_setup.md` @@ -32,6 +45,7 @@ - Existing Vuexy-based assets (see `docs/index.md`) pending catalog alignment. ### Database (PostgreSQL) +- **Assessment Item Bank Models** — Persistent item bank, response items, and exposure stats backing adaptive selection. → `./docs/functions/assessment_item_bank_models.md` - Core assessment schema coverage documented in `docs/Database/database_info.md`. ### Infra/Provisioning diff --git a/docs/erd.mmd b/docs/erd.mmd index 268a40b..bb54bdb 100644 --- a/docs/erd.mmd +++ b/docs/erd.mmd @@ -1,5 +1,5 @@ erDiagram - SCORE_TEMPLATE ||--o{ SCORE_BUCKET : "qualifies" - SCORE_TEMPLATE ||--o{ SCORE_DIMENSION : "weights" - SCORE_DIMENSION ||--o{ SCORE_ITEM : "covers" - SCORE_TEMPLATE }o--|| SCORE_SCALE : "uses" + ASSESSMENT_VERSIONS ||--o{ ASSESSMENT_ITEM_BANK : "contains" + ASSESSMENT_ITEM_BANK ||--|| ASSESSMENT_ITEM_STATS : "tracks" + ASSESSMENTS ||--o{ ASSESSMENT_RESPONSE_ITEMS : "records" + ASSESSMENT_ITEM_BANK ||--o{ ASSESSMENT_RESPONSE_ITEMS : "delivers" diff --git a/docs/functions/assessment_blueprint_engine.md b/docs/functions/assessment_blueprint_engine.md new file mode 100644 index 0000000..f63430f --- /dev/null +++ b/docs/functions/assessment_blueprint_engine.md @@ -0,0 +1,91 @@ +--- +langs: [en, nb-NO] +lastUpdated: 2025-09-24 +--- + +# Assessment Blueprint Engine — Overview + +**en:** Stratified blueprint loader and selector powering adaptive assessment forms. Sources: `src/app/assessment_engine/blueprint_engine.py`, `src/app/assessment_engine/__init__.py`, `src/app/schemas/assessment_blueprint.py`, `src/app/api/v1/assessments.py`, `tests/test_assessment_blueprint_engine.py`. + +**nb-NO:** Stratifisert blueprint-laster og trekker for adaptive vurderingsskjema. Kildebaner: `src/app/assessment_engine/blueprint_engine.py`, `src/app/assessment_engine/__init__.py`, `src/app/schemas/assessment_blueprint.py`, `src/app/api/v1/assessments.py`, `tests/test_assessment_blueprint_engine.py`. + +**SPOT:** ./SPOT.md#function-catalog + +## API + +- `GET /api/v1/assessment/blueprint` – Lists bundled blueprint summaries for operators. +- `GET /api/v1/assessment/blueprint/{template_id}/preview` – Returns deterministic item selection previews (optional `seed`). +- Python helpers: + - `load_blueprint_document(template_id)` – Parse blueprint JSON into validated model. + - `select_items(document, pool, seen_codes=None, rng=None)` – Stratified sampling respecting quotas, anchors, exposure caps. + - `score_responses(selected_items, item_scores, document)` – Weighted dimension scoring with knockout policies. + - `generate_selection_preview(template_id, seed=None)` – Utility harness for API exposure/tests. + - `pool_from_item_bank(records, stats=None)` – Merge persisted item bank rows with live stats before selection. + +## Design + +- Blueprint definitions live in `src/app/assessment_templates/blueprints/*.json` and follow the `AssessmentBlueprintDocument` schema (quotas, anchors, critical policies, scoring, sample pools). +- Selection honours three layers: + 1. Explicit anchor items → subtract matching difficulty quotas. + 2. Remaining per-dimension anchor quota. + 3. Difficulty buckets with weighted random sampling (difficulty factors, discrimination, exposure penalty with default cap fallback). +- Exposure caps default to blueprint `exposure.default_cap` unless item overrides. +- Scoring aggregates dimension scores with configurable weight strategies and supports two critical modes: knockout (bucket override) or weighted (multiplied weight). +- Preview endpoints reuse bundled sample pools so API consumers can verify quotas without hitting production item banks. +- `pool_from_item_bank` hydrates `BlueprintItem` objects from the Postgres item bank, preferring live discrimination/exposure metrics when stats exist. +- Error handling surfaces invalid/missing blueprint packages with `BlueprintLoadError` mapped to HTTP 404. + +## Usage + +```python +from app.assessment_engine import load_blueprint_document, sample_pool_from_blueprint, select_items, score_responses + +document = load_blueprint_document("course_gdpr_social_email_cookies_v1_no") +pool = sample_pool_from_blueprint(document) +selected = select_items(document, pool, seen_codes={"safety_easy_intro_rules"}) +responses = {item.code: 0.8 for item in selected} +summary = score_responses(selected, responses, document) +print(summary.overall_bucket) +``` + +When sourcing items from the persistent bank: + +```python +from app.assessment_engine import pool_from_item_bank + +pool = pool_from_item_bank(db_items, db_stats) +selected = select_items(document, pool, seen_codes=session_history) +``` + +HTTP preview example: + +```bash +curl "http://localhost:8000/api/v1/assessment/blueprint/course_gdpr_social_email_cookies_v1_no/preview?seed=42" +``` + +## Changelog + +### [Unreleased] +- 2025-09-24: Initial blueprint loader, selector, scoring engine, and API previews. +- 2025-09-24: Added item bank hydration helper aligning selection with persisted exposure statistics. + +## Diagrams + +```mermaid +digraph BlueprintFlow { + rankdir=LR + BlueprintJSON["Blueprint JSON\n(quotas, anchors)"] + SamplePool["Sample Pool\n(items with stats)"] + Selector["Stratified Selector"] + Responses["Responses\n(normalised 0..1)"] + Scorer["Scoring Engine\n(weights + knockout)"] + Summary["Dimension & Total Summary"] + + BlueprintJSON -> Selector + SamplePool -> Selector + Selector -> Responses + Responses -> Scorer + Selector -> Scorer + Scorer -> Summary +} +``` diff --git a/docs/functions/assessment_item_bank_models.md b/docs/functions/assessment_item_bank_models.md new file mode 100644 index 0000000..9fffb6d --- /dev/null +++ b/docs/functions/assessment_item_bank_models.md @@ -0,0 +1,72 @@ +--- +langs: [en, nb-NO] +lastUpdated: 2025-09-24 +--- + +# Assessment Item Bank Models — Overview + +**en:** SQLAlchemy models and Postgres schema powering blueprint-governed item banks, response linkage, and exposure statistics. Sources: `src/app/models/assessment_item_bank.py`, `seed_scripts/03_assessment_item_bank_schema.sql`. + +**nb-NO:** SQLAlchemy-modeller og Postgres-skjema for blueprint-styrt spørsmålsbank, responskobling og eksponeringsstatistikk. Kildebaner: `src/app/models/assessment_item_bank.py`, `seed_scripts/03_assessment_item_bank_schema.sql`. + +**SPOT:** ./SPOT.md#function-catalog + +## API + +- SQLAlchemy models: + - `AssessmentVersion` — Links template IDs with blueprint names/versions. + - `AssessmentItemBank` — Stores per-item difficulty, weights, anchors, and metadata. + - `AssessmentResponseItem` — Persists rendered items, answers, and item-level scores per assessment attempt. + - `AssessmentItemStats` — Tracks facility, discrimination, and exposure for rotation governance. +- Schema bootstrap: `seed_scripts/03_assessment_item_bank_schema.sql` creates tables, constraints, and indexes idempotently. + +## Design + +- Item lifecycle is scoped by `assessment_versions` so blueprint releases can rotate independently of assessments. +- JSONB columns hold tags, arbitrary metadata, and submitted answers for flexible reporting. +- Cascading deletes clear bank entries when a version is retired while response rows retain referential integrity via `RESTRICT`. +- Stats table centralises KPI inputs (facility, discrimination, exposure) used by the selector to prioritise low-exposure/high-signal items. +- Exposure caps are stored per item with optional overrides; selection helper merges real-time stats before sampling. + +## Usage + +```python +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.assessment_engine import pool_from_item_bank +from app.models.assessment_item_bank import AssessmentItemBank, AssessmentItemStats + + +async def load_pool(session: AsyncSession, version_id): + records = ( + await session.execute( + select(AssessmentItemBank).where(AssessmentItemBank.version_id == version_id) + ) + ).scalars().all() + + stats = ( + await session.execute( + select(AssessmentItemStats).where(AssessmentItemStats.item_id.in_([r.item_id for r in records])) + ) + ).scalars().all() + + return pool_from_item_bank(records, stats) +``` + +**nb-NO:** Kjør `seed_scripts/03_assessment_item_bank_schema.sql` før bruk for å sikre tabellene finnes. + +## Changelog + +### [Unreleased] +- 2025-09-24: Introduced assessment version catalog, item bank, response link, and stats models + schema script. + +## Diagrams + +```mermaid +erDiagram + ASSESSMENT_VERSIONS ||--o{ ASSESSMENT_ITEM_BANK : "contains" + ASSESSMENT_ITEM_BANK ||--|| ASSESSMENT_ITEM_STATS : "tracks" + ASSESSMENTS ||--o{ ASSESSMENT_RESPONSE_ITEMS : "records" + ASSESSMENT_ITEM_BANK ||--o{ ASSESSMENT_RESPONSE_ITEMS : "delivers" +``` diff --git a/docs/schema.md b/docs/schema.md index 12abf70..1ab7c80 100644 --- a/docs/schema.md +++ b/docs/schema.md @@ -5,9 +5,9 @@ lastUpdated: 2025-09-24 # Data Model Overview -**en:** Simplified entity diagram for score templates and related concepts. +**en:** Conceptual view of assessment versions, item bank governance, response linkage, and exposure statistics. -**nb-NO:** Forenklet enhetsdiagram for skårmaler og tilhørende begreper. +**nb-NO:** Konseptuelt diagram over vurderingsversjoner, spørsmålsbank-styring, responskobling og eksponeringsstatistikk. ![ERD](../public/erd.svg) @@ -16,10 +16,10 @@ lastUpdated: 2025-09-24 ```mermaid erDiagram - SCORE_TEMPLATE ||--o{ SCORE_BUCKET : "qualifies" - SCORE_TEMPLATE ||--o{ SCORE_DIMENSION : "weights" - SCORE_DIMENSION ||--o{ SCORE_ITEM : "covers" - SCORE_TEMPLATE }o--|| SCORE_SCALE : "uses" + ASSESSMENT_VERSIONS ||--o{ ASSESSMENT_ITEM_BANK : "contains" + ASSESSMENT_ITEM_BANK ||--|| ASSESSMENT_ITEM_STATS : "tracks" + ASSESSMENTS ||--o{ ASSESSMENT_RESPONSE_ITEMS : "records" + ASSESSMENT_ITEM_BANK ||--o{ ASSESSMENT_RESPONSE_ITEMS : "delivers" ``` diff --git a/docs/tasks/tasklist.md b/docs/tasks/tasklist.md index fffe9e0..998ee18 100644 --- a/docs/tasks/tasklist.md +++ b/docs/tasks/tasklist.md @@ -11,3 +11,7 @@ - [x] Verify diagrams exist for `api_router_v1` (Diagrams section) - [x] Update docs & SPOT for `core_setup`; CHANGELOG [Unreleased] - [x] Verify diagrams exist for `core_setup` (Diagrams section) +- [x] Update docs & SPOT for `assessment_blueprint_engine`; CHANGELOG [Unreleased] +- [x] Verify diagrams exist for `assessment_blueprint_engine` (Diagrams section) +- [x] Update docs & SPOT for `assessment_item_bank_models`; CHANGELOG [Unreleased] +- [x] Verify diagrams exist for `assessment_item_bank_models` (Diagrams section) diff --git a/public/erd.svg b/public/erd.svg index 438c567..102d17a 100644 --- a/public/erd.svg +++ b/public/erd.svg @@ -1,28 +1,34 @@ - - Score Template ERD (Simplified) - Conceptual relationship between score templates, buckets, dimensions, items, and scale. + + Conceptual relationship between assessment versions, item bank records, response linkage, and exposure statistics. - + - - SCORE_TEMPLATE - - SCORE_BUCKET - - SCORE_DIMENSION - - SCORE_ITEM - - SCORE_SCALE - - - - + + ASSESSMENT_VERSIONS + + + ASSESSMENT_ITEM_BANK + + + ASSESSMENT_ITEM_STATS + + + ASSESSMENTS + + + ASSESSMENT_RESPONSE + ITEMS + + + + + + diff --git a/seed_scripts/03_assessment_item_bank_schema.sql b/seed_scripts/03_assessment_item_bank_schema.sql new file mode 100644 index 0000000..0349e31 --- /dev/null +++ b/seed_scripts/03_assessment_item_bank_schema.sql @@ -0,0 +1,74 @@ +-- Docs: ./docs/functions/assessment_item_bank_models.md +-- SPOT: ./SPOT.md#function-catalog + +-- Ensure UUID generation helpers are available +CREATE EXTENSION IF NOT EXISTS "pgcrypto"; + +-- Catalog of assessment versions and blueprint releases +CREATE TABLE IF NOT EXISTS assessment_versions ( + version_id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + template_id text NOT NULL, + blueprint_name text NOT NULL, + blueprint_version text NOT NULL, + notes text, + is_active boolean NOT NULL DEFAULT true, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz, + CONSTRAINT uq_assessment_versions_template_blueprint + UNIQUE (template_id, blueprint_name, blueprint_version) +); + +CREATE INDEX IF NOT EXISTS idx_assessment_versions_template + ON assessment_versions (template_id); + +-- Item bank entries governed by blueprint quotas +CREATE TABLE IF NOT EXISTS assessment_item_bank ( + item_id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + version_id uuid NOT NULL REFERENCES assessment_versions(version_id) ON DELETE CASCADE, + code text NOT NULL, + dimension text NOT NULL, + difficulty text NOT NULL CHECK (difficulty IN ('easy','medium','hard')), + weight numeric NOT NULL DEFAULT 1.0, + critical boolean NOT NULL DEFAULT false, + anchor boolean NOT NULL DEFAULT false, + discrimination numeric, + exposure_cap numeric, + tags jsonb NOT NULL DEFAULT '[]'::jsonb, + meta jsonb, + created_at timestamptz NOT NULL DEFAULT now(), + CONSTRAINT uq_assessment_item_bank_version_code UNIQUE (version_id, code) +); + +CREATE INDEX IF NOT EXISTS idx_assessment_item_bank_dimension + ON assessment_item_bank (dimension); +CREATE INDEX IF NOT EXISTS idx_assessment_item_bank_difficulty + ON assessment_item_bank (difficulty); +CREATE INDEX IF NOT EXISTS idx_assessment_item_bank_anchor + ON assessment_item_bank (anchor); + +-- Exposure and performance statistics per item +CREATE TABLE IF NOT EXISTS assessment_item_stats ( + item_id uuid PRIMARY KEY REFERENCES assessment_item_bank(item_id) ON DELETE CASCADE, + shown integer NOT NULL DEFAULT 0, + correct integer NOT NULL DEFAULT 0, + facility numeric, + discrimination numeric, + exposure numeric, + last_seen_at timestamptz +); + +CREATE INDEX IF NOT EXISTS idx_assessment_item_stats_exposure + ON assessment_item_stats (exposure); + +-- Captured responses to individual blueprint-driven items +CREATE TABLE IF NOT EXISTS assessment_response_items ( + assessment_id uuid NOT NULL REFERENCES assessments(id) ON DELETE CASCADE, + item_id uuid NOT NULL REFERENCES assessment_item_bank(item_id) ON DELETE RESTRICT, + answer jsonb, + score numeric, + responded_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (assessment_id, item_id) +); + +CREATE INDEX IF NOT EXISTS idx_assessment_response_items_item + ON assessment_response_items (item_id); diff --git a/src/app/api/v1/assessments.py b/src/app/api/v1/assessments.py index 6532887..7eb8160 100644 --- a/src/app/api/v1/assessments.py +++ b/src/app/api/v1/assessments.py @@ -1,16 +1,23 @@ +"""Docs: ./docs/functions/assessment_blueprint_engine.md | SPOT: ./SPOT.md#function-catalog""" import secrets import uuid from datetime import datetime from typing import Annotated, List from uuid import UUID -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.api.dependencies import get_current_user from ...core.db.database import async_get_db +from ...assessment_engine import ( + BlueprintLoadError, + generate_selection_preview, + list_blueprint_ids, + load_blueprint_document, +) from ...assessment_templates import ( get_assessment_template, list_assessment_templates, @@ -33,10 +40,71 @@ AssessmentDefinition, AssessmentTemplateSummary, ) +from ...schemas.assessment_blueprint import ( + BlueprintPreviewItem, + BlueprintPreviewResponse, + BlueprintSummary, +) router = APIRouter() + + +@router.get("/blueprint", response_model=list[BlueprintSummary]) +async def list_blueprint_documents() -> list[BlueprintSummary]: + """Expose available assessment blueprints with quota metadata.""" + + summaries: list[BlueprintSummary] = [] + for blueprint_id in list_blueprint_ids(): + try: + document = load_blueprint_document(blueprint_id) + except BlueprintLoadError as exc: + raise HTTPException(status_code=500, detail=str(exc)) from exc + summaries.append( + BlueprintSummary( + template_id=document.template_id, + blueprint_name=document.blueprint_name, + version=document.version, + anchors=document.anchors, + total_quota=document.total_quota, + ) + ) + return summaries + + +@router.get("/blueprint/{template_id}/preview", response_model=BlueprintPreviewResponse) +async def preview_blueprint_selection(template_id: str, seed: int | None = Query(default=None, ge=0)) -> BlueprintPreviewResponse: + """Return a deterministic preview for stratified selection per blueprint.""" + + try: + preview = generate_selection_preview(template_id, seed) + document = load_blueprint_document(template_id) + except BlueprintLoadError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except Exception as exc: # pragma: no cover - defensive + raise HTTPException(status_code=500, detail=f"Failed to prepare blueprint preview: {exc}") from exc + + template_summary = None + if preview.get("template"): + template_summary = AssessmentTemplateSummary(**preview["template"]) + + blueprint_summary = BlueprintSummary( + template_id=document.template_id, + blueprint_name=document.blueprint_name, + version=document.version, + anchors=document.anchors, + total_quota=document.total_quota, + ) + items = [BlueprintPreviewItem.model_validate(item) for item in preview.get("selected_items", [])] + + return BlueprintPreviewResponse( + template=template_summary.model_dump() if template_summary else None, + blueprint=blueprint_summary, + items=items, + ) + + @router.get("/schema", response_model=list[AssessmentTemplateSummary]) async def list_available_assessment_templates() -> list[AssessmentTemplateSummary]: """List all bundled assessment templates.""" diff --git a/src/app/assessment_engine/__init__.py b/src/app/assessment_engine/__init__.py new file mode 100644 index 0000000..996175d --- /dev/null +++ b/src/app/assessment_engine/__init__.py @@ -0,0 +1,28 @@ +"""Docs: ./docs/functions/assessment_blueprint_engine.md | SPOT: ./SPOT.md#function-catalog""" +from .blueprint_engine import ( + BlueprintItem, + BlueprintLoadError, + DimensionScore, + ScoreSummary, + generate_selection_preview, + list_blueprint_ids, + load_blueprint_document, + pool_from_item_bank, + sample_pool_from_blueprint, + score_responses, + select_items, +) + +__all__ = [ + "BlueprintItem", + "BlueprintLoadError", + "DimensionScore", + "ScoreSummary", + "generate_selection_preview", + "list_blueprint_ids", + "load_blueprint_document", + "pool_from_item_bank", + "sample_pool_from_blueprint", + "score_responses", + "select_items", +] diff --git a/src/app/assessment_engine/blueprint_engine.py b/src/app/assessment_engine/blueprint_engine.py new file mode 100644 index 0000000..c286438 --- /dev/null +++ b/src/app/assessment_engine/blueprint_engine.py @@ -0,0 +1,478 @@ +"""Docs: ./docs/functions/assessment_blueprint_engine.md | SPOT: ./SPOT.md#function-catalog""" +from __future__ import annotations + +import random +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Iterable, Mapping, MutableMapping, Sequence +from uuid import UUID + +from pydantic import ValidationError + +from app.assessment_templates import get_assessment_template +from app.schemas.assessment_template import AssessmentTemplateSummary +from app.schemas.assessment_blueprint import ( + AssessmentBlueprintDocument, + BucketThreshold, + CriticalDimensionPolicy, + DimensionBlueprint, + DimensionScoringPolicy, + DifficultyWeights, + SampleItem, +) + +if TYPE_CHECKING: + from app.models.assessment_item_bank import ( + AssessmentItemBank, + AssessmentItemStats, + ) + +try: # Python 3.11+ + from importlib import resources +except ImportError: # pragma: no cover - fallback for legacy interpreters + import importlib_resources as resources # type: ignore[assignment] + + +BLUEPRINT_PACKAGE = "app.assessment_templates.blueprints" + + +@dataclass(slots=True) +class BlueprintItem: + """Runtime item representation with scoring metadata.""" + + code: str + dimension: str + difficulty: str + weight: float = 1.0 + is_anchor: bool = False + is_critical: bool = False + discrimination: float | None = None + exposure_ratio: float | None = None + exposure_cap: float | None = None + tags: tuple[str, ...] = field(default_factory=tuple) + + def selection_weight(self, weights: DifficultyWeights, min_weight: float) -> float: + """Compute a sampling weight honoring difficulty, exposure, and discrimination.""" + + difficulty_factor = getattr(weights, self.difficulty, 1.0) + discrimination = self.discrimination if self.discrimination and self.discrimination > 0 else 1.0 + exposure_penalty = 1.0 + if self.exposure_ratio is not None: + exposure_penalty = max(min_weight, 1.0 - self.exposure_ratio) + return max(min_weight, difficulty_factor * discrimination * exposure_penalty) + + +class BlueprintLoadError(RuntimeError): + """Raised when blueprint resources cannot be loaded or parsed.""" + + +def load_blueprint_document(template_id: str) -> AssessmentBlueprintDocument: + """Load and validate the blueprint document associated with a template.""" + + base_path = resources.files(BLUEPRINT_PACKAGE) + try: + blueprint_path = base_path.joinpath(f"{template_id}.json") + except FileNotFoundError as exc: # pragma: no cover - defensive + raise BlueprintLoadError(f"Blueprint package missing for {template_id}") from exc + + if not blueprint_path.is_file(): + raise BlueprintLoadError(f"No blueprint available for template '{template_id}'") + + raw = blueprint_path.read_text(encoding="utf-8") + try: + return AssessmentBlueprintDocument.model_validate_json(raw) + except ValidationError as exc: # pragma: no cover - surfaces configuration issues clearly + raise BlueprintLoadError(f"Invalid blueprint for template '{template_id}': {exc}") from exc + + +def list_blueprint_ids() -> list[str]: + """Return all blueprint identifiers bundled with the application.""" + + base_path = resources.files(BLUEPRINT_PACKAGE) + return [path.stem for path in base_path.iterdir() if path.suffix == ".json"] + + +def _as_item(sample: SampleItem, blueprint: AssessmentBlueprintDocument) -> BlueprintItem: + """Convert a sample item specification into the runtime structure.""" + + is_critical = sample.critical or sample.code in blueprint.critical.items + dimension_policy = blueprint.critical.dimensions.get(sample.dimension) + if dimension_policy and dimension_policy.mode == "knockout": + is_critical = True + return BlueprintItem( + code=sample.code, + dimension=sample.dimension, + difficulty=sample.difficulty, + weight=sample.weight, + is_anchor=sample.anchor or sample.code in {anchor.code for anchor in blueprint.anchor_items}, + is_critical=is_critical or sample.code in blueprint.dimensions.get(sample.dimension, DimensionBlueprint()).critical_items, + discrimination=sample.discrimination, + exposure_ratio=sample.exposure_ratio, + exposure_cap=sample.exposure_cap if sample.exposure_cap is not None else blueprint.exposure.default_cap, + tags=tuple(sample.tags), + ) + + +def pool_from_item_bank( + bank_items: Sequence["AssessmentItemBank"], + stats: Mapping["UUID", "AssessmentItemStats"] | Sequence["AssessmentItemStats"] | None = None, +) -> list[BlueprintItem]: + """Hydrate blueprint items from persisted bank records and optional statistics.""" + + stats_lookup: dict[UUID, AssessmentItemStats] = {} + if stats: + if isinstance(stats, Mapping): + stats_lookup = dict(stats) + else: + stats_lookup = {entry.item_id: entry for entry in stats} + + hydrated: list[BlueprintItem] = [] + for record in bank_items: + stat = stats_lookup.get(record.item_id) + discrimination = None + exposure_ratio = None + + if stat and stat.discrimination is not None: + discrimination = float(stat.discrimination) + elif record.discrimination is not None: + discrimination = float(record.discrimination) + + if stat and stat.exposure is not None: + exposure_ratio = max(0.0, min(1.0, float(stat.exposure))) + elif record.meta and isinstance(record.meta, dict): + maybe_exposure = record.meta.get("exposure_ratio") + if isinstance(maybe_exposure, (int, float)): + exposure_ratio = max(0.0, min(1.0, float(maybe_exposure))) + + tags_value = () + if isinstance(record.tags, (list, tuple)): + tags_value = tuple(str(tag) for tag in record.tags) + + hydrated.append( + BlueprintItem( + code=record.code, + dimension=record.dimension, + difficulty=record.difficulty, + weight=float(record.weight) if record.weight is not None else 1.0, + is_anchor=bool(record.anchor), + is_critical=bool(record.critical), + discrimination=discrimination, + exposure_ratio=exposure_ratio, + exposure_cap=float(record.exposure_cap) + if record.exposure_cap is not None + else None, + tags=tags_value, + ) + ) + + return hydrated + + +def sample_pool_from_blueprint(document: AssessmentBlueprintDocument) -> list[BlueprintItem]: + """Create a pool of items using the inline sample metadata if present.""" + + return [_as_item(sample, document) for sample in document.sample_pool] + + +def _filter_exposure(items: list[BlueprintItem]) -> list[BlueprintItem]: + """Filter out items that exceeded their exposure cap, keeping fallbacks.""" + + if not items: + return items + limited = [ + item + for item in items + if item.exposure_cap is None + or item.exposure_ratio is None + or item.exposure_ratio < item.exposure_cap + ] + return limited or items + + +def _weighted_sample( + rng: random.Random, + candidates: list[BlueprintItem], + count: int, + difficulty_weights: DifficultyWeights, + min_weight: float, +) -> list[BlueprintItem]: + """Select items without replacement using proportional weights.""" + + if count <= 0 or not candidates: + return [] + + working = candidates.copy() + selected: list[BlueprintItem] = [] + for _ in range(min(count, len(working))): + weights = [item.selection_weight(difficulty_weights, min_weight) for item in working] + total = sum(weights) + if total <= 0: + choice_index = rng.randrange(len(working)) + else: + pick = rng.random() * total + cumulative = 0.0 + choice_index = 0 + for idx, weight in enumerate(weights): + cumulative += weight + if cumulative >= pick: + choice_index = idx + break + selected.append(working.pop(choice_index)) + return selected + + +def _subtract_quota(quota: dict[str, int], difficulty: str) -> None: + if difficulty in quota and quota[difficulty] > 0: + quota[difficulty] -= 1 + + +def select_items( + document: AssessmentBlueprintDocument, + pool: Sequence[BlueprintItem], + seen_codes: Iterable[str] | None = None, + rng: random.Random | None = None, +) -> list[BlueprintItem]: + """Select items following the blueprint quotas, anchors, and exposure rules.""" + + rng = rng or random.Random() + seen = set(seen_codes or []) + available = [item for item in pool if item.code not in seen] + selected: list[BlueprintItem] = [] + quota_remaining: dict[str, dict[str, int]] = { + dim: {"easy": spec.easy, "medium": spec.medium, "hard": spec.hard} + for dim, spec in document.dimensions.items() + } + + # 1) Explicit anchor items + anchor_lookup = {anchor.code: anchor for anchor in document.anchor_items} + for anchor_code, anchor in anchor_lookup.items(): + match = next((item for item in available if item.code == anchor_code), None) + if not match: + continue + selected.append(match) + available.remove(match) + difficulty = anchor.difficulty or match.difficulty + _subtract_quota(quota_remaining.setdefault(match.dimension, {}), difficulty) + + # 2) Additional anchors per dimension quota + for dimension, spec in document.dimensions.items(): + required = max(0, spec.anchors - sum(1 for item in selected if item.dimension == dimension and item.is_anchor)) + if required == 0: + continue + candidates = [ + item + for item in available + if item.dimension == dimension and item.is_anchor + ] + candidates = _filter_exposure(candidates) + picks = _weighted_sample(rng, candidates, required, document.difficulty_weights, document.exposure.min_weight) + for item in picks: + selected.append(item) + available.remove(item) + _subtract_quota(quota_remaining.setdefault(item.dimension, {}), item.difficulty) + + # 3) General selection per dimension/difficulty + for dimension, difficulty_quota in quota_remaining.items(): + for difficulty, count in difficulty_quota.items(): + if count <= 0: + continue + candidates = [ + item + for item in available + if item.dimension == dimension and item.difficulty == difficulty and not item.is_anchor + ] + candidates = _filter_exposure(candidates) + picks = _weighted_sample(rng, candidates, count, document.difficulty_weights, document.exposure.min_weight) + for item in picks: + selected.append(item) + available.remove(item) + remaining = count - len(picks) + if remaining > 0: + # Fallback: any remaining difficulty within the dimension + fallback_candidates = [ + item + for item in available + if item.dimension == dimension and not item.is_anchor + ] + fallback_candidates = _filter_exposure(fallback_candidates) + fallback_picks = _weighted_sample( + rng, + fallback_candidates, + remaining, + document.difficulty_weights, + document.exposure.min_weight, + ) + for item in fallback_picks: + selected.append(item) + available.remove(item) + + return selected + + +@dataclass +class DimensionScore: + code: str + raw_score: float + max_score: float + percentage: float + bucket: str + knockout_triggered: bool + + +@dataclass +class ScoreSummary: + overall_score: float + overall_bucket: str + dimensions: list[DimensionScore] + + +def _bucket_for_value(thresholds: Sequence[BucketThreshold], value: float) -> str: + for threshold in thresholds: + if value >= threshold.min: + return threshold.code + return thresholds[-1].code if thresholds else "RED" + + +def score_responses( + selected_items: Sequence[BlueprintItem], + item_scores: Mapping[str, float], + document: AssessmentBlueprintDocument, +) -> ScoreSummary: + """Aggregate weighted scores per dimension and apply knockout logic.""" + + item_map: MutableMapping[str, BlueprintItem] = {item.code: item for item in selected_items} + dimension_policies: Mapping[str, DimensionScoringPolicy] = { + policy.code: policy for policy in document.scoring.dimensions + } + critical_policy: Mapping[str, CriticalDimensionPolicy] = document.critical.dimensions + dimension_results: list[DimensionScore] = [] + knockout_dimensions: set[str] = set() + + for dimension, policy in dimension_policies.items(): + items = [item for item in selected_items if item.dimension == dimension] + if not items: + dimension_results.append( + DimensionScore( + code=dimension, + raw_score=0.0, + max_score=0.0, + percentage=0.0, + bucket=document.scoring.buckets[-1].code, + knockout_triggered=False, + ) + ) + continue + + critical_conf = critical_policy.get(dimension) + raw_score = 0.0 + max_score = 0.0 + knockout_triggered = False + + for item in items: + response = item_scores.get(item.code, 0.0) + response = max(0.0, min(1.0, response)) + if policy.item_weight_mode == "equal": + item_weight = 1.0 + else: + item_weight = item.weight + + if critical_conf and item.is_critical and critical_conf.mode == "weighted": + item_weight *= critical_conf.weight_multiplier + elif item.is_critical: + item_weight *= policy.critical_weight_multiplier + + raw_score += response * item_weight + max_score += item_weight + + if ( + policy.critical_knockout + and item.is_critical + and response < policy.critical_threshold + ): + knockout_triggered = True + + if ( + critical_conf + and critical_conf.mode == "knockout" + and item.is_critical + and response < critical_conf.threshold + ): + knockout_triggered = True + + percentage = (raw_score / max_score * 100) if max_score > 0 else 0.0 + bucket = _bucket_for_value(document.scoring.buckets, percentage) + if knockout_triggered: + bucket = document.scoring.overall_knockout_bucket + knockout_dimensions.add(dimension) + dimension_results.append( + DimensionScore( + code=dimension, + raw_score=raw_score, + max_score=max_score, + percentage=percentage, + bucket=bucket, + knockout_triggered=knockout_triggered, + ) + ) + + overall_score = 0.0 + for result in dimension_results: + policy = dimension_policies[result.code] + contribution = result.percentage * policy.weight + overall_score += contribution + + overall_bucket = _bucket_for_value(document.scoring.buckets, overall_score) + if knockout_dimensions: + overall_bucket = document.scoring.overall_knockout_bucket + + return ScoreSummary( + overall_score=overall_score, + overall_bucket=overall_bucket, + dimensions=dimension_results, + ) + + +def generate_selection_preview(template_id: str, seed: int | None = None) -> dict[str, object]: + """Utility helper for API exposure: run a deterministic sample draw.""" + + blueprint = load_blueprint_document(template_id) + rng = random.Random(seed or 42) + pool = sample_pool_from_blueprint(blueprint) + selected = select_items(blueprint, pool, rng=rng) + template = get_assessment_template(template_id) + summary = AssessmentTemplateSummary.from_definition(template) if template else None + return { + "template": summary.model_dump() if summary else None, + "blueprint": { + "template_id": blueprint.template_id, + "name": blueprint.blueprint_name, + "version": blueprint.version, + "anchors": blueprint.anchors, + "total_quota": blueprint.total_quota, + }, + "selected_items": [ + { + "code": item.code, + "dimension": item.dimension, + "difficulty": item.difficulty, + "weight": item.weight, + "anchor": item.is_anchor, + "critical": item.is_critical, + } + for item in selected + ], + } + + +__all__ = [ + "BlueprintItem", + "BlueprintLoadError", + "DimensionScore", + "ScoreSummary", + "generate_selection_preview", + "list_blueprint_ids", + "load_blueprint_document", + "pool_from_item_bank", + "sample_pool_from_blueprint", + "score_responses", + "select_items", +] diff --git a/src/app/assessment_templates/blueprints/course_gdpr_social_email_cookies_v1_no.json b/src/app/assessment_templates/blueprints/course_gdpr_social_email_cookies_v1_no.json new file mode 100644 index 0000000..075c859 --- /dev/null +++ b/src/app/assessment_templates/blueprints/course_gdpr_social_email_cookies_v1_no.json @@ -0,0 +1,375 @@ +{ + "template_id": "course_gdpr_social_email_cookies_v1_no", + "blueprint_name": "Safety Culture & Leadership Form A", + "version": "2025-09-24", + "anchors": 2, + "dimensions": { + "SAFETY": { + "easy": 2, + "medium": 4, + "hard": 2, + "anchors": 1, + "critical_items": [ + "safety_anchor_ppe", + "safety_hard_confined" + ] + }, + "CULTURE": { + "easy": 1, + "medium": 4, + "hard": 1, + "anchors": 1, + "critical_items": [ + "culture_anchor_voice" + ] + }, + "LEADERSHIP": { + "easy": 1, + "medium": 4, + "hard": 1, + "anchors": 0, + "critical_items": [ + "leadership_hard_accountability" + ] + } + }, + "anchor_items": [ + { + "code": "safety_anchor_ppe", + "dimension": "SAFETY", + "difficulty": "medium" + }, + { + "code": "culture_anchor_voice", + "dimension": "CULTURE", + "difficulty": "medium" + } + ], + "difficulty_weights": { + "easy": 1.0, + "medium": 1.2, + "hard": 1.5 + }, + "exposure": { + "default_cap": 0.25, + "min_weight": 0.05 + }, + "critical": { + "items": [ + "safety_anchor_ppe", + "safety_hard_confined", + "culture_anchor_voice" + ], + "dimensions": { + "SAFETY": { + "mode": "knockout", + "threshold": 0.5, + "weight_multiplier": 3.0 + }, + "CULTURE": { + "mode": "weighted", + "threshold": 0.6, + "weight_multiplier": 3.0 + }, + "LEADERSHIP": { + "mode": "weighted", + "threshold": 0.6, + "weight_multiplier": 2.5 + } + } + }, + "scoring": { + "dimensions": [ + { + "code": "SAFETY", + "weight": 0.4, + "critical_knockout": true, + "critical_threshold": 0.5, + "item_weight_mode": "use_item_weight", + "critical_weight_multiplier": 3.0 + }, + { + "code": "CULTURE", + "weight": 0.3, + "critical_knockout": false, + "critical_threshold": 0.5, + "item_weight_mode": "use_item_weight", + "critical_weight_multiplier": 3.0 + }, + { + "code": "LEADERSHIP", + "weight": 0.3, + "critical_knockout": false, + "critical_threshold": 0.5, + "item_weight_mode": "use_item_weight", + "critical_weight_multiplier": 2.5 + } + ], + "buckets": [ + { + "code": "GREEN", + "min": 80 + }, + { + "code": "YELLOW", + "min": 65 + }, + { + "code": "ORANGE", + "min": 45 + }, + { + "code": "RED", + "min": 0 + } + ], + "overall_knockout_bucket": "RED" + }, + "sample_pool": [ + { + "code": "safety_anchor_ppe", + "dimension": "SAFETY", + "difficulty": "medium", + "weight": 4.0, + "anchor": true, + "critical": true, + "discrimination": 0.42, + "exposure_ratio": 0.12, + "tags": [ + "PPE", + "anchor" + ] + }, + { + "code": "safety_medium_incident_reporting", + "dimension": "SAFETY", + "difficulty": "medium", + "weight": 3.5, + "critical": true, + "discrimination": 0.38, + "exposure_ratio": 0.18, + "tags": [ + "incident", + "reporting" + ] + }, + { + "code": "safety_medium_toolbox_talks", + "dimension": "SAFETY", + "difficulty": "medium", + "weight": 3.0, + "discrimination": 0.25, + "exposure_ratio": 0.05, + "tags": [ + "culture", + "communication" + ] + }, + { + "code": "safety_medium_shift_handover", + "dimension": "SAFETY", + "difficulty": "medium", + "weight": 3.2, + "discrimination": 0.3, + "exposure_ratio": 0.21, + "tags": [ + "handover" + ] + }, + { + "code": "safety_easy_intro_rules", + "dimension": "SAFETY", + "difficulty": "easy", + "weight": 2.0, + "discrimination": 0.22, + "exposure_ratio": 0.08, + "tags": [ + "onboarding" + ] + }, + { + "code": "safety_easy_microlearning", + "dimension": "SAFETY", + "difficulty": "easy", + "weight": 2.2, + "discrimination": 0.19, + "exposure_ratio": 0.09, + "tags": [ + "training" + ] + }, + { + "code": "safety_hard_confined", + "dimension": "SAFETY", + "difficulty": "hard", + "weight": 4.5, + "critical": true, + "discrimination": 0.5, + "exposure_ratio": 0.24, + "tags": [ + "confined", + "space" + ] + }, + { + "code": "safety_hard_electrical", + "dimension": "SAFETY", + "difficulty": "hard", + "weight": 4.3, + "discrimination": 0.41, + "exposure_ratio": 0.2, + "tags": [ + "lockout", + "tagout" + ] + }, + { + "code": "culture_anchor_voice", + "dimension": "CULTURE", + "difficulty": "medium", + "weight": 3.3, + "anchor": true, + "critical": true, + "discrimination": 0.36, + "exposure_ratio": 0.14, + "tags": [ + "speakup", + "anchor" + ] + }, + { + "code": "culture_medium_peer_feedback", + "dimension": "CULTURE", + "difficulty": "medium", + "weight": 3.1, + "discrimination": 0.28, + "exposure_ratio": 0.17, + "tags": [ + "feedback" + ] + }, + { + "code": "culture_medium_learning_loops", + "dimension": "CULTURE", + "difficulty": "medium", + "weight": 3.0, + "discrimination": 0.26, + "exposure_ratio": 0.11, + "tags": [ + "learning" + ] + }, + { + "code": "culture_medium_conflict_resolution", + "dimension": "CULTURE", + "difficulty": "medium", + "weight": 3.4, + "discrimination": 0.3, + "exposure_ratio": 0.23, + "tags": [ + "conflict" + ] + }, + { + "code": "culture_medium_values_activation", + "dimension": "CULTURE", + "difficulty": "medium", + "weight": 3.2, + "discrimination": 0.27, + "exposure_ratio": 0.2, + "tags": [ + "values" + ] + }, + { + "code": "culture_easy_rituals", + "dimension": "CULTURE", + "difficulty": "easy", + "weight": 2.4, + "discrimination": 0.21, + "exposure_ratio": 0.1, + "tags": [ + "rituals" + ] + }, + { + "code": "culture_hard_retention", + "dimension": "CULTURE", + "difficulty": "hard", + "weight": 4.0, + "critical": true, + "discrimination": 0.4, + "exposure_ratio": 0.19, + "tags": [ + "retention" + ] + }, + { + "code": "leadership_easy_checkins", + "dimension": "LEADERSHIP", + "difficulty": "easy", + "weight": 2.5, + "discrimination": 0.2, + "exposure_ratio": 0.07, + "tags": [ + "checkins" + ] + }, + { + "code": "leadership_medium_coaching", + "dimension": "LEADERSHIP", + "difficulty": "medium", + "weight": 3.4, + "discrimination": 0.32, + "exposure_ratio": 0.16, + "tags": [ + "coaching" + ] + }, + { + "code": "leadership_medium_vision", + "dimension": "LEADERSHIP", + "difficulty": "medium", + "weight": 3.3, + "discrimination": 0.29, + "exposure_ratio": 0.15, + "tags": [ + "vision" + ] + }, + { + "code": "leadership_medium_resources", + "dimension": "LEADERSHIP", + "difficulty": "medium", + "weight": 3.5, + "discrimination": 0.31, + "exposure_ratio": 0.18, + "tags": [ + "resources" + ] + }, + { + "code": "leadership_medium_retros", + "dimension": "LEADERSHIP", + "difficulty": "medium", + "weight": 3.2, + "discrimination": 0.24, + "exposure_ratio": 0.2, + "tags": [ + "retrospective" + ] + }, + { + "code": "leadership_hard_accountability", + "dimension": "LEADERSHIP", + "difficulty": "hard", + "weight": 4.1, + "critical": true, + "discrimination": 0.37, + "exposure_ratio": 0.22, + "tags": [ + "accountability" + ] + } + ] +} diff --git a/src/app/models/__init__.py b/src/app/models/__init__.py index c44c105..d32178d 100644 --- a/src/app/models/__init__.py +++ b/src/app/models/__init__.py @@ -1,4 +1,21 @@ +from .assessment_item_bank import ( + AssessmentItemBank, + AssessmentItemStats, + AssessmentResponseItem, + AssessmentVersion, +) from .post import Post from .rate_limit import RateLimit from .tier import Tier from .user import User + +__all__ = [ + "AssessmentItemBank", + "AssessmentItemStats", + "AssessmentResponseItem", + "AssessmentVersion", + "Post", + "RateLimit", + "Tier", + "User", +] diff --git a/src/app/models/assessment_item_bank.py b/src/app/models/assessment_item_bank.py new file mode 100644 index 0000000..029fed7 --- /dev/null +++ b/src/app/models/assessment_item_bank.py @@ -0,0 +1,137 @@ +"""Docs: ./docs/functions/assessment_item_bank_models.md | SPOT: ./SPOT.md#function-catalog""" +from __future__ import annotations + +import uuid + +from sqlalchemy import ( + Boolean, + CheckConstraint, + Column, + DateTime, + ForeignKey, + Index, + Integer, + Numeric, + String, + Text, + UniqueConstraint, + text, +) +from sqlalchemy.dialects.postgresql import JSONB, UUID +from sqlalchemy.sql import func + +from app.core.db.database import Base + + +class AssessmentVersion(Base): + """Catalog entry linking templates and blueprint releases.""" + + __tablename__ = "assessment_versions" + + version_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + template_id = Column(String(128), nullable=False) + blueprint_name = Column(String(128), nullable=False) + blueprint_version = Column(String(64), nullable=False) + notes = Column(Text, nullable=True) + is_active = Column(Boolean, nullable=False, default=True) + created_at = Column(DateTime(timezone=True), server_default=func.now()) + updated_at = Column(DateTime(timezone=True), onupdate=func.now()) + + __table_args__ = ( + UniqueConstraint( + "template_id", + "blueprint_name", + "blueprint_version", + name="uq_assessment_versions_template_blueprint", + ), + Index("idx_assessment_versions_template", "template_id"), + ) + + +class AssessmentItemBank(Base): + """Persisted item metadata for adaptive selection routines.""" + + __tablename__ = "assessment_item_bank" + + item_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + version_id = Column( + UUID(as_uuid=True), + ForeignKey("assessment_versions.version_id", ondelete="CASCADE"), + nullable=False, + ) + code = Column(String(128), nullable=False) + dimension = Column(String(64), nullable=False) + difficulty = Column(String(16), nullable=False) + weight = Column(Numeric(8, 4), nullable=False, server_default=text("1.0")) + critical = Column(Boolean, nullable=False, server_default=text("false")) + anchor = Column(Boolean, nullable=False, server_default=text("false")) + discrimination = Column(Numeric(8, 4), nullable=True) + exposure_cap = Column(Numeric(6, 4), nullable=True) + tags = Column(JSONB, nullable=False, server_default=text("'[]'::jsonb")) + meta = Column(JSONB, nullable=True) + created_at = Column(DateTime(timezone=True), server_default=func.now()) + + __table_args__ = ( + CheckConstraint( + "difficulty IN ('easy','medium','hard')", + name="chk_assessment_item_bank_difficulty", + ), + UniqueConstraint("version_id", "code", name="uq_assessment_item_bank_version_code"), + Index("idx_assessment_item_bank_dimension", "dimension"), + Index("idx_assessment_item_bank_difficulty", "difficulty"), + Index("idx_assessment_item_bank_anchor", "anchor"), + ) + + +class AssessmentItemStats(Base): + """Aggregated usage statistics for blueprint items.""" + + __tablename__ = "assessment_item_stats" + + item_id = Column( + UUID(as_uuid=True), + ForeignKey("assessment_item_bank.item_id", ondelete="CASCADE"), + primary_key=True, + ) + shown = Column(Integer, nullable=False, server_default=text("0")) + correct = Column(Integer, nullable=False, server_default=text("0")) + facility = Column(Numeric(6, 4), nullable=True) + discrimination = Column(Numeric(6, 4), nullable=True) + exposure = Column(Numeric(6, 4), nullable=True) + last_seen_at = Column(DateTime(timezone=True), nullable=True) + + __table_args__ = ( + Index("idx_assessment_item_stats_exposure", "exposure"), + ) + + +class AssessmentResponseItem(Base): + """Link table storing rendered items and participant scoring.""" + + __tablename__ = "assessment_response_items" + + assessment_id = Column( + UUID(as_uuid=True), + ForeignKey("assessments.id", ondelete="CASCADE"), + primary_key=True, + ) + item_id = Column( + UUID(as_uuid=True), + ForeignKey("assessment_item_bank.item_id", ondelete="RESTRICT"), + primary_key=True, + ) + answer = Column(JSONB, nullable=True) + score = Column(Numeric(6, 4), nullable=True) + responded_at = Column(DateTime(timezone=True), server_default=func.now()) + + __table_args__ = ( + Index("idx_assessment_response_items_item", "item_id"), + ) + + +__all__ = [ + "AssessmentItemBank", + "AssessmentItemStats", + "AssessmentResponseItem", + "AssessmentVersion", +] diff --git a/src/app/schemas/assessment_blueprint.py b/src/app/schemas/assessment_blueprint.py new file mode 100644 index 0000000..4ec40ea --- /dev/null +++ b/src/app/schemas/assessment_blueprint.py @@ -0,0 +1,284 @@ +"""Docs: ./docs/functions/assessment_blueprint_engine.md | SPOT: ./SPOT.md#function-catalog""" +from __future__ import annotations + +from typing import Dict, List, Literal + +from enum import Enum + +from pydantic import BaseModel, ConfigDict, Field, RootModel, model_validator + +from app.schemas.assessment_template import AssessmentTemplateSummary + +ItemDifficulty = Literal["easy", "medium", "hard"] + +class DifficultyQuota(BaseModel): + """Requested number of items per difficulty bucket.""" + + easy: int = Field(default=0, ge=0) + medium: int = Field(default=0, ge=0) + hard: int = Field(default=0, ge=0) + + model_config = ConfigDict(extra="forbid") + + @property + def total(self) -> int: + return self.easy + self.medium + self.hard + + +class AnchorItem(BaseModel): + """Explicit anchor allocation for deterministic inclusion.""" + + code: str = Field(..., description="Stable item identifier.") + dimension: str = Field(..., description="Blueprint dimension this anchor belongs to.") + difficulty: ItemDifficulty | None = Field( + default=None, + description="Optional hint to decrement the correct difficulty quota.", + ) + + model_config = ConfigDict(extra="forbid") + + +class DimensionBlueprint(BaseModel): + """Blueprint quotas and metadata for one assessment dimension.""" + + easy: int = Field(default=0, ge=0) + medium: int = Field(default=0, ge=0) + hard: int = Field(default=0, ge=0) + anchors: int = Field(default=0, ge=0) + critical_items: List[str] = Field(default_factory=list) + + model_config = ConfigDict(extra="forbid") + + @property + def quota(self) -> DifficultyQuota: + return DifficultyQuota(easy=self.easy, medium=self.medium, hard=self.hard) + + @model_validator(mode="after") + def validate_anchor_quota(self) -> "DimensionBlueprint": + if self.anchors > self.quota.total: + raise ValueError("anchor quota cannot exceed total quota for dimension") + return self + + +class CriticalMode(str, Enum): + """Supported critical handling strategies.""" + + KNOCKOUT = "knockout" + WEIGHTED = "weighted" + + +class CriticalDimensionPolicy(BaseModel): + """Critical handling rules for a dimension.""" + + mode: CriticalMode = Field(default=CriticalMode.WEIGHTED) + threshold: float = Field(default=0.5, ge=0.0, le=1.0) + weight_multiplier: float = Field(default=3.0, ge=1.0) + + model_config = ConfigDict(extra="forbid") + + +class CriticalBlueprint(BaseModel): + """Global critical item and dimension policy definitions.""" + + items: List[str] = Field(default_factory=list) + dimensions: Dict[str, CriticalDimensionPolicy] = Field(default_factory=dict) + + model_config = ConfigDict(extra="forbid") + + +class BucketThreshold(BaseModel): + """Threshold definition for qualitative score buckets.""" + + code: str + min: float = Field(..., ge=0, le=100) + + model_config = ConfigDict(extra="forbid") + + +class DimensionScoringPolicy(BaseModel): + """Scoring behaviour for an individual dimension.""" + + code: str + weight: float = Field(..., gt=0) + critical_knockout: bool = False + critical_threshold: float = Field(default=0.5, ge=0.0, le=1.0) + item_weight_mode: Literal["use_item_weight", "equal"] = "use_item_weight" + critical_weight_multiplier: float = Field(default=3.0, ge=1.0) + + model_config = ConfigDict(extra="forbid") + + +class ScoringBlueprint(BaseModel): + """Aggregation and bucket calculation rules.""" + + dimensions: List[DimensionScoringPolicy] + buckets: List[BucketThreshold] = Field( + default_factory=lambda: [ + BucketThreshold(code="GREEN", min=80), + BucketThreshold(code="YELLOW", min=60), + BucketThreshold(code="ORANGE", min=40), + BucketThreshold(code="RED", min=0), + ] + ) + overall_knockout_bucket: str = "RED" + + model_config = ConfigDict(extra="forbid") + + @model_validator(mode="after") + def validate_weights(self) -> "ScoringBlueprint": + total = sum(d.weight for d in self.dimensions) + if abs(total - 1.0) > 1e-6: + raise ValueError("dimension weights must sum to 1.0") + return self + + @model_validator(mode="after") + def validate_bucket_order(self) -> "ScoringBlueprint": + mins = [bucket.min for bucket in self.buckets] + if mins != sorted(mins, reverse=True): + raise ValueError("bucket thresholds must be provided in descending order") + return self + + +class DifficultyWeights(BaseModel): + """Optional weighting factors per difficulty for selection priority.""" + + easy: float = Field(default=1.0, ge=0.0) + medium: float = Field(default=1.2, ge=0.0) + hard: float = Field(default=1.4, ge=0.0) + + model_config = ConfigDict(extra="forbid") + + +class ExposureSettings(BaseModel): + """Global exposure defaults for selection balancing.""" + + default_cap: float | None = Field(default=0.25, ge=0.0, le=1.0) + min_weight: float = Field(default=0.05, ge=0.0) + + model_config = ConfigDict(extra="forbid") + + +class SampleItem(BaseModel): + """Inline sample item metadata to exercise the blueprint logic.""" + + code: str + dimension: str + difficulty: ItemDifficulty + weight: float = 1.0 + anchor: bool = False + critical: bool = False + discrimination: float | None = None + exposure_ratio: float | None = Field(default=None, ge=0.0, le=1.0) + exposure_cap: float | None = Field(default=None, ge=0.0, le=1.0) + tags: List[str] = Field(default_factory=list) + + model_config = ConfigDict(extra="forbid") + + +class AssessmentBlueprintDocument(BaseModel): + """Complete blueprint specification for an assessment version.""" + + template_id: str + blueprint_name: str + version: str + anchors: int = Field(default=0, ge=0) + dimensions: Dict[str, DimensionBlueprint] + anchor_items: List[AnchorItem] = Field(default_factory=list) + difficulty_weights: DifficultyWeights = Field(default_factory=DifficultyWeights) + exposure: ExposureSettings = Field(default_factory=ExposureSettings) + critical: CriticalBlueprint = Field(default_factory=CriticalBlueprint) + scoring: ScoringBlueprint + sample_pool: List[SampleItem] = Field(default_factory=list) + + model_config = ConfigDict(extra="forbid") + + @property + def total_quota(self) -> int: + return sum(d.quota.total for d in self.dimensions.values()) + + @model_validator(mode="after") + def validate_anchor_counts(self) -> "AssessmentBlueprintDocument": + dimension_anchor_total = sum(d.anchors for d in self.dimensions.values()) + anchor_item_total = len(self.anchor_items) + expected = max(dimension_anchor_total, anchor_item_total) + if expected > self.anchors: + raise ValueError( + "declared anchors must cover explicit per-dimension and listed anchor items" + ) + return self + + @model_validator(mode="after") + def validate_dimension_alignment(self) -> "AssessmentBlueprintDocument": + dimension_codes = set(self.dimensions.keys()) + scoring_codes = {rule.code for rule in self.scoring.dimensions} + missing = dimension_codes.symmetric_difference(scoring_codes) + if missing: + raise ValueError( + "blueprint and scoring dimensions must match exactly; mismatches: " + + ", ".join(sorted(missing)) + ) + critical_codes = set(self.critical.dimensions.keys()) + invalid_critical = critical_codes - dimension_codes + if invalid_critical: + raise ValueError( + "critical dimension policies reference unknown dimensions: " + + ", ".join(sorted(invalid_critical)) + ) + return self + + +BlueprintRegistry = RootModel[Dict[str, AssessmentBlueprintDocument]] + + +class BlueprintSummary(BaseModel): + """Lightweight metadata representation for blueprint listings.""" + + template_id: str + blueprint_name: str + version: str + anchors: int + total_quota: int + + model_config = ConfigDict(extra="forbid") + + +class BlueprintPreviewItem(BaseModel): + """Selected item payload used in previews and testing.""" + + code: str + dimension: str + difficulty: ItemDifficulty + weight: float + anchor: bool + critical: bool + + model_config = ConfigDict(extra="forbid") + + +class BlueprintPreviewResponse(BaseModel): + """Preview payload bundling template, blueprint, and item info.""" + + template: AssessmentTemplateSummary | None + blueprint: BlueprintSummary + items: List[BlueprintPreviewItem] + + model_config = ConfigDict(extra="forbid", arbitrary_types_allowed=True) + + +__all__ = [ + "AssessmentBlueprintDocument", + "BlueprintPreviewItem", + "BlueprintPreviewResponse", + "BlueprintSummary", + "AnchorItem", + "BlueprintRegistry", + "CriticalBlueprint", + "CriticalDimensionPolicy", + "DimensionBlueprint", + "DimensionScoringPolicy", + "DifficultyQuota", + "DifficultyWeights", + "ExposureSettings", + "SampleItem", + "ScoringBlueprint", +] diff --git a/tests/conftest.py b/tests/conftest.py index 935cef9..2f1b4f6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -10,8 +10,8 @@ from sqlalchemy.orm import sessionmaker from sqlalchemy.orm.session import Session -from src.app.core.config import settings -from src.app.main import app +from app.core.config import settings +from app.main import app DATABASE_URI = settings.POSTGRES_URI DATABASE_PREFIX = settings.POSTGRES_SYNC_PREFIX @@ -74,7 +74,7 @@ def sample_user_read(): """Generate a sample UserRead object.""" import uuid - from src.app.schemas.user import UserRead + from app.schemas.user import UserRead return UserRead( id=1, diff --git a/tests/helpers/generators.py b/tests/helpers/generators.py index 7ba73d0..006679c 100644 --- a/tests/helpers/generators.py +++ b/tests/helpers/generators.py @@ -2,8 +2,8 @@ from sqlalchemy.orm import Session -from src.app import models -from src.app.core.security import get_password_hash +from app import models +from app.core.security import get_password_hash from tests.conftest import fake diff --git a/tests/helpers/mocks.py b/tests/helpers/mocks.py index 713ae68..12792a6 100644 --- a/tests/helpers/mocks.py +++ b/tests/helpers/mocks.py @@ -2,7 +2,7 @@ from fastapi.encoders import jsonable_encoder -from src.app import models +from app import models from tests.conftest import fake diff --git a/tests/test_assessment_blueprint_engine.py b/tests/test_assessment_blueprint_engine.py new file mode 100644 index 0000000..610b062 --- /dev/null +++ b/tests/test_assessment_blueprint_engine.py @@ -0,0 +1,116 @@ +"""Docs: ./docs/functions/assessment_blueprint_engine.md | SPOT: ./SPOT.md#function-catalog""" +from __future__ import annotations + +import random +from decimal import Decimal +from uuid import uuid4 + +from app.assessment_engine import ( + ScoreSummary, + load_blueprint_document, + pool_from_item_bank, + sample_pool_from_blueprint, + score_responses, + select_items, +) +from app.schemas.assessment_blueprint import BlueprintPreviewItem +from app.models.assessment_item_bank import AssessmentItemBank, AssessmentItemStats + +BLUEPRINT_ID = "course_gdpr_social_email_cookies_v1_no" + + +def test_blueprint_document_metadata() -> None: + document = load_blueprint_document(BLUEPRINT_ID) + assert document.blueprint_name == "Safety Culture & Leadership Form A" + assert document.anchors == 2 + assert document.total_quota == 20 + + +def test_select_items_matches_blueprint() -> None: + document = load_blueprint_document(BLUEPRINT_ID) + pool = sample_pool_from_blueprint(document) + rng = random.Random(2024) + selected = select_items(document, pool, rng=rng) + + assert len(selected) == document.total_quota + safety_items = [item for item in selected if item.dimension == "SAFETY"] + culture_items = [item for item in selected if item.dimension == "CULTURE"] + leadership_items = [item for item in selected if item.dimension == "LEADERSHIP"] + + assert len(safety_items) == 8 + assert len(culture_items) == 6 + assert len(leadership_items) == 6 + + anchors = [item for item in selected if item.is_anchor] + assert {item.code for item in anchors} >= {"safety_anchor_ppe", "culture_anchor_voice"} + + safety_by_difficulty = {bucket: sum(1 for item in safety_items if item.difficulty == bucket) for bucket in {"easy", "medium", "hard"}} + assert safety_by_difficulty["easy"] == 2 + assert safety_by_difficulty["medium"] == 4 + assert safety_by_difficulty["hard"] == 2 + + +def test_score_responses_applies_knockout() -> None: + document = load_blueprint_document(BLUEPRINT_ID) + pool = sample_pool_from_blueprint(document) + rng = random.Random(17) + selected = select_items(document, pool, rng=rng) + + item_scores = {item.code: 0.9 for item in selected} + item_scores["safety_anchor_ppe"] = 0.2 # Trigger knockout on safety + summary: ScoreSummary = score_responses(selected, item_scores, document) + + safety_dimension = next(dim for dim in summary.dimensions if dim.code == "SAFETY") + assert safety_dimension.knockout_triggered is True + assert summary.overall_bucket == document.scoring.overall_knockout_bucket + + +def test_pool_from_item_bank_prefers_stats_over_record() -> None: + version_id = uuid4() + item_id = uuid4() + record = AssessmentItemBank() + record.item_id = item_id + record.version_id = version_id + record.code = "safety_anchor_ppe" + record.dimension = "SAFETY" + record.difficulty = "medium" + record.weight = Decimal("2.5") + record.critical = True + record.anchor = True + record.discrimination = Decimal("0.40") + record.exposure_cap = Decimal("0.25") + record.tags = ["PPE", "anchor"] + record.meta = {"exposure_ratio": 0.18} + stats = AssessmentItemStats() + stats.item_id = item_id + stats.shown = 25 + stats.correct = 18 + stats.facility = Decimal("0.72") + stats.discrimination = Decimal("0.55") + stats.exposure = Decimal("0.32") + + hydrated = pool_from_item_bank([record], [stats]) + assert len(hydrated) == 1 + item = hydrated[0] + assert item.code == "safety_anchor_ppe" + assert item.is_anchor is True + assert item.is_critical is True + assert item.discrimination == 0.55 + assert item.exposure_ratio == 0.32 + assert item.exposure_cap == 0.25 + assert item.tags == ("PPE", "anchor") + + +def test_blueprint_preview_endpoint(client) -> None: + response = client.get("/api/v1/assessment/blueprint") + assert response.status_code == 200 + payload = response.json() + assert any(item["template_id"] == BLUEPRINT_ID for item in payload) + + preview_response = client.get(f"/api/v1/assessment/blueprint/{BLUEPRINT_ID}/preview", params={"seed": 99}) + assert preview_response.status_code == 200 + preview_payload = preview_response.json() + assert preview_payload["blueprint"]["template_id"] == BLUEPRINT_ID + assert len(preview_payload["items"]) == 20 + first_item = BlueprintPreviewItem.model_validate(preview_payload["items"][0]) + assert first_item.dimension in {"SAFETY", "CULTURE", "LEADERSHIP"} diff --git a/tests/test_user.py b/tests/test_user.py index b5ef269..5951d6d 100644 --- a/tests/test_user.py +++ b/tests/test_user.py @@ -4,9 +4,9 @@ import pytest -from src.app.api.v1.users import erase_user, patch_user, read_user, read_users, write_user -from src.app.core.exceptions.http_exceptions import DuplicateValueException, ForbiddenException, NotFoundException -from src.app.schemas.user import UserCreate, UserRead, UserUpdate +from app.api.v1.users import erase_user, patch_user, read_user, read_users, write_user +from app.core.exceptions.http_exceptions import DuplicateValueException, ForbiddenException, NotFoundException +from app.schemas.user import UserCreate, UserRead, UserUpdate class TestWriteUser: @@ -17,13 +17,13 @@ async def test_create_user_success(self, mock_db, sample_user_data, sample_user_ """Test successful user creation.""" user_create = UserCreate(**sample_user_data) - with patch("src.app.api.v1.users.crud_users") as mock_crud: + with patch("app.api.v1.users.crud_users") as mock_crud: # Mock that email and username don't exist mock_crud.exists = AsyncMock(side_effect=[False, False]) # email, then username mock_crud.create = AsyncMock(return_value=Mock(id=1)) mock_crud.get = AsyncMock(return_value=sample_user_read) - with patch("src.app.api.v1.users.get_password_hash") as mock_hash: + with patch("app.api.v1.users.get_password_hash") as mock_hash: mock_hash.return_value = "hashed_password" result = await write_user(Mock(), user_create, mock_db) @@ -38,7 +38,7 @@ async def test_create_user_duplicate_email(self, mock_db, sample_user_data): """Test user creation with duplicate email.""" user_create = UserCreate(**sample_user_data) - with patch("src.app.api.v1.users.crud_users") as mock_crud: + with patch("app.api.v1.users.crud_users") as mock_crud: # Mock that email already exists mock_crud.exists = AsyncMock(return_value=True) @@ -50,7 +50,7 @@ async def test_create_user_duplicate_username(self, mock_db, sample_user_data): """Test user creation with duplicate username.""" user_create = UserCreate(**sample_user_data) - with patch("src.app.api.v1.users.crud_users") as mock_crud: + with patch("app.api.v1.users.crud_users") as mock_crud: # Mock email doesn't exist, but username does mock_crud.exists = AsyncMock(side_effect=[False, True]) @@ -66,7 +66,7 @@ async def test_read_user_success(self, mock_db, sample_user_read): """Test successful user retrieval.""" username = "test_user" - with patch("src.app.api.v1.users.crud_users") as mock_crud: + with patch("app.api.v1.users.crud_users") as mock_crud: mock_crud.get = AsyncMock(return_value=sample_user_read) result = await read_user(Mock(), username, mock_db) @@ -81,7 +81,7 @@ async def test_read_user_not_found(self, mock_db): """Test user retrieval when user doesn't exist.""" username = "nonexistent_user" - with patch("src.app.api.v1.users.crud_users") as mock_crud: + with patch("app.api.v1.users.crud_users") as mock_crud: mock_crud.get = AsyncMock(return_value=None) with pytest.raises(NotFoundException, match="User not found"): @@ -96,10 +96,10 @@ async def test_read_users_success(self, mock_db): """Test successful users list retrieval.""" mock_users_data = {"data": [{"id": 1}, {"id": 2}], "count": 2} - with patch("src.app.api.v1.users.crud_users") as mock_crud: + with patch("app.api.v1.users.crud_users") as mock_crud: mock_crud.get_multi = AsyncMock(return_value=mock_users_data) - with patch("src.app.api.v1.users.paginated_response") as mock_paginated: + with patch("app.api.v1.users.paginated_response") as mock_paginated: expected_response = {"data": [{"id": 1}, {"id": 2}], "pagination": {}} mock_paginated.return_value = expected_response @@ -120,7 +120,7 @@ async def test_patch_user_success(self, mock_db, current_user_dict, sample_user_ sample_user_read.username = username # Make sure usernames match user_update = UserUpdate(name="New Name") - with patch("src.app.api.v1.users.crud_users") as mock_crud: + with patch("app.api.v1.users.crud_users") as mock_crud: mock_crud.get = AsyncMock(return_value=sample_user_read) mock_crud.exists = AsyncMock(return_value=False) # No conflicts mock_crud.update = AsyncMock(return_value=None) @@ -137,7 +137,7 @@ async def test_patch_user_forbidden(self, mock_db, current_user_dict, sample_use sample_user_read.username = username user_update = UserUpdate(name="New Name") - with patch("src.app.api.v1.users.crud_users") as mock_crud: + with patch("app.api.v1.users.crud_users") as mock_crud: mock_crud.get = AsyncMock(return_value=sample_user_read) with pytest.raises(ForbiddenException): @@ -154,11 +154,11 @@ async def test_erase_user_success(self, mock_db, current_user_dict, sample_user_ sample_user_read.username = username token = "mock_token" - with patch("src.app.api.v1.users.crud_users") as mock_crud: + with patch("app.api.v1.users.crud_users") as mock_crud: mock_crud.get = AsyncMock(return_value=sample_user_read) mock_crud.delete = AsyncMock(return_value=None) - with patch("src.app.api.v1.users.blacklist_token", new_callable=AsyncMock) as mock_blacklist: + with patch("app.api.v1.users.blacklist_token", new_callable=AsyncMock) as mock_blacklist: result = await erase_user(Mock(), username, current_user_dict, mock_db, token) assert result == {"message": "User deleted"} @@ -171,7 +171,7 @@ async def test_erase_user_not_found(self, mock_db, current_user_dict): username = "nonexistent_user" token = "mock_token" - with patch("src.app.api.v1.users.crud_users") as mock_crud: + with patch("app.api.v1.users.crud_users") as mock_crud: mock_crud.get = AsyncMock(return_value=None) with pytest.raises(NotFoundException, match="User not found"): @@ -184,7 +184,7 @@ async def test_erase_user_forbidden(self, mock_db, current_user_dict, sample_use sample_user_read.username = username token = "mock_token" - with patch("src.app.api.v1.users.crud_users") as mock_crud: + with patch("app.api.v1.users.crud_users") as mock_crud: mock_crud.get = AsyncMock(return_value=sample_user_read) with pytest.raises(ForbiddenException):