276 lines
9.7 KiB
Python
276 lines
9.7 KiB
Python
"""Phase 3 KPI evidence export helpers.
|
|
|
|
This module turns persisted pre/post aggregate scores into the Phase 3 evidence
|
|
shape checked by scripts/check-phase3-artifacts.py. It does not claim clinical
|
|
effectiveness; it only produces pilot evidence files for operator review.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import json
|
|
from collections import defaultdict
|
|
from datetime import UTC, date, datetime
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from typing import Any, Iterable, Mapping, Sequence
|
|
from uuid import UUID
|
|
|
|
PREPOST_MEASURE_NAMES = (
|
|
"self_efficacy",
|
|
"skill_proficiency",
|
|
"training_satisfaction",
|
|
)
|
|
PREPOST_TIMEPOINTS = ("pre", "post")
|
|
PHASE3_KPI_METRICS = (
|
|
"embedding_consistency",
|
|
"hallucination_rate",
|
|
"icc",
|
|
"inter_rater_kappa",
|
|
"pilot_completion",
|
|
"self_efficacy_prepost",
|
|
"session_completion",
|
|
"sus",
|
|
"top1",
|
|
)
|
|
PREPOST_CSV_PATH = "02-measures/prepost_measures.csv"
|
|
KPI_REPORT_PATH = "02-measures/kpi_report.json"
|
|
|
|
|
|
class ParticipantKeys:
|
|
def __init__(self) -> None:
|
|
self._keys: dict[str, str] = {}
|
|
|
|
def key(self, raw_id: Any) -> str:
|
|
value = str(raw_id or "unknown-participant")
|
|
if value not in self._keys:
|
|
self._keys[value] = f"P3-{len(self._keys) + 1:03d}"
|
|
return self._keys[value]
|
|
|
|
def __len__(self) -> int:
|
|
return len(self._keys)
|
|
|
|
|
|
def json_safe(value: Any) -> Any:
|
|
if isinstance(value, datetime):
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=UTC)
|
|
return value.isoformat().replace("+00:00", "Z")
|
|
if isinstance(value, date):
|
|
return value.isoformat()
|
|
if isinstance(value, (Decimal, UUID)):
|
|
return str(value)
|
|
if isinstance(value, Mapping):
|
|
return {str(key): json_safe(item) for key, item in value.items()}
|
|
if isinstance(value, list):
|
|
return [json_safe(item) for item in value]
|
|
if isinstance(value, tuple):
|
|
return [json_safe(item) for item in value]
|
|
return value
|
|
|
|
|
|
def iso_timestamp(value: Any) -> str:
|
|
safe = json_safe(value)
|
|
return safe if isinstance(safe, str) else str(safe or "")
|
|
|
|
|
|
def normalized_score(raw_score: float, min_score: float, max_score: float) -> float:
|
|
if max_score <= min_score:
|
|
return 0.0
|
|
return round(((raw_score - min_score) / (max_score - min_score)) * 100.0, 3)
|
|
|
|
|
|
def latest_prepost_rows(rows: Iterable[Mapping[str, Any]]) -> list[dict[str, Any]]:
|
|
latest: dict[tuple[str, str, str], dict[str, Any]] = {}
|
|
for row in rows:
|
|
learner_id = str(row.get("learner_id") or row.get("participant_id") or "")
|
|
measure_name = str(row.get("measure_name") or "")
|
|
timepoint = str(row.get("timepoint") or "")
|
|
if measure_name not in PREPOST_MEASURE_NAMES or timepoint not in PREPOST_TIMEPOINTS:
|
|
continue
|
|
key = (learner_id, measure_name, timepoint)
|
|
current = dict(row)
|
|
current_order = iso_timestamp(current.get("updated_at") or current.get("collected_at"))
|
|
previous = latest.get(key)
|
|
previous_order = iso_timestamp(previous.get("updated_at") or previous.get("collected_at")) if previous else ""
|
|
if previous is None or current_order >= previous_order:
|
|
latest[key] = current
|
|
return sorted(
|
|
latest.values(),
|
|
key=lambda item: (
|
|
str(item.get("learner_id") or item.get("participant_id") or ""),
|
|
str(item.get("measure_name") or ""),
|
|
str(item.get("timepoint") or ""),
|
|
),
|
|
)
|
|
|
|
|
|
def build_prepost_csv_rows(
|
|
rows: Iterable[Mapping[str, Any]],
|
|
*,
|
|
participant_keys: ParticipantKeys | None = None,
|
|
) -> list[dict[str, str]]:
|
|
keys = participant_keys if participant_keys is not None else ParticipantKeys()
|
|
output: list[dict[str, str]] = []
|
|
for row in latest_prepost_rows(rows):
|
|
raw_score = float(row.get("raw_score") or row.get("score") or 0.0)
|
|
output.append(
|
|
{
|
|
"participant_id": keys.key(row.get("learner_id") or row.get("participant_id")),
|
|
"measure_name": str(row.get("measure_name") or ""),
|
|
"timepoint": str(row.get("timepoint") or ""),
|
|
"score": _format_number(raw_score),
|
|
"collected_at": iso_timestamp(row.get("collected_at") or row.get("updated_at")),
|
|
}
|
|
)
|
|
return output
|
|
|
|
|
|
def paired_prepost_summary(rows: Iterable[Mapping[str, Any]], measure_name: str) -> dict[str, Any]:
|
|
by_participant: dict[str, dict[str, float]] = defaultdict(dict)
|
|
for row in latest_prepost_rows(rows):
|
|
if str(row.get("measure_name") or "") != measure_name:
|
|
continue
|
|
participant_id = str(row.get("learner_id") or row.get("participant_id") or "")
|
|
raw_score = float(row.get("raw_score") or row.get("score") or 0.0)
|
|
min_score = float(row.get("min_score") or 1.0)
|
|
max_score = float(row.get("max_score") or 5.0)
|
|
by_participant[participant_id][str(row.get("timepoint") or "")] = normalized_score(
|
|
raw_score,
|
|
min_score,
|
|
max_score,
|
|
)
|
|
|
|
deltas: list[float] = []
|
|
pre_values: list[float] = []
|
|
post_values: list[float] = []
|
|
missing_pairs = 0
|
|
for values in by_participant.values():
|
|
if "pre" not in values or "post" not in values:
|
|
missing_pairs += 1
|
|
continue
|
|
pre_values.append(values["pre"])
|
|
post_values.append(values["post"])
|
|
deltas.append(values["post"] - values["pre"])
|
|
|
|
return {
|
|
"participants_with_any_measure": len(by_participant),
|
|
"complete_pairs": len(deltas),
|
|
"missing_pairs": missing_pairs,
|
|
"mean_pre": _mean(pre_values),
|
|
"mean_post": _mean(post_values),
|
|
"mean_delta": _mean(deltas),
|
|
}
|
|
|
|
|
|
def build_kpi_report(
|
|
rows: Sequence[Mapping[str, Any]],
|
|
*,
|
|
pilot_id: str,
|
|
generated_at: str,
|
|
source_window: Mapping[str, Any] | None = None,
|
|
review_operator: str = "",
|
|
) -> dict[str, Any]:
|
|
latest_rows = latest_prepost_rows(rows)
|
|
participants = {str(row.get("learner_id") or row.get("participant_id") or "") for row in latest_rows}
|
|
metrics = {name: _placeholder_metric(name) for name in PHASE3_KPI_METRICS}
|
|
|
|
self_efficacy = paired_prepost_summary(latest_rows, "self_efficacy")
|
|
metrics["self_efficacy_prepost"] = {
|
|
"value": self_efficacy["mean_delta"],
|
|
"threshold": 0.0,
|
|
"pass": False,
|
|
"numerator": self_efficacy["complete_pairs"],
|
|
"denominator": max(self_efficacy["participants_with_any_measure"], 0),
|
|
"method": "paired normalized post-pre delta for pilot review; no official pass/fail gate",
|
|
"source_files": [PREPOST_CSV_PATH],
|
|
"mean_pre": self_efficacy["mean_pre"],
|
|
"mean_post": self_efficacy["mean_post"],
|
|
"mean_delta": self_efficacy["mean_delta"],
|
|
"complete_pairs": self_efficacy["complete_pairs"],
|
|
"missing_pairs": self_efficacy["missing_pairs"],
|
|
}
|
|
|
|
for measure_name in ("skill_proficiency", "training_satisfaction"):
|
|
summary = paired_prepost_summary(latest_rows, measure_name)
|
|
metrics[f"{measure_name}_prepost"] = {
|
|
**_placeholder_metric(f"{measure_name}_prepost"),
|
|
**summary,
|
|
"value": summary["mean_delta"],
|
|
"numerator": summary["complete_pairs"],
|
|
"denominator": summary["participants_with_any_measure"],
|
|
"method": "paired normalized post-pre delta for pilot review; not a required KPI gate yet",
|
|
"source_files": [PREPOST_CSV_PATH],
|
|
}
|
|
|
|
return {
|
|
"pilot_id": pilot_id,
|
|
"generated_at": generated_at,
|
|
"source_window": dict(source_window or _source_window(latest_rows)),
|
|
"cohort_size": len(participants),
|
|
"metrics": metrics,
|
|
"exclusions": [],
|
|
"open_schema_gaps": [
|
|
"official item text and validated scoring rules are not encoded here",
|
|
"experimental/control assignment and statistical testing require owner/evaluation-design approval",
|
|
],
|
|
"review": {
|
|
"operator": review_operator,
|
|
"reviewed_at": "",
|
|
"decision": "pending",
|
|
},
|
|
}
|
|
|
|
|
|
def write_prepost_csv(rows: Sequence[Mapping[str, str]], path: Path) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", encoding="utf-8", newline="") as handle:
|
|
writer = csv.DictWriter(
|
|
handle,
|
|
fieldnames=("participant_id", "measure_name", "timepoint", "score", "collected_at"),
|
|
)
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
|
|
def write_kpi_report(report: Mapping[str, Any], path: Path) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
json.dumps(json_safe(report), ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _placeholder_metric(name: str) -> dict[str, Any]:
|
|
return {
|
|
"value": 0.0,
|
|
"threshold": 0.0,
|
|
"pass": False,
|
|
"numerator": 0,
|
|
"denominator": 0,
|
|
"method": f"not computed by prepost export scaffold: {name}",
|
|
"source_files": [],
|
|
}
|
|
|
|
|
|
def _source_window(rows: Sequence[Mapping[str, Any]]) -> dict[str, str]:
|
|
timestamps = [
|
|
iso_timestamp(row.get("collected_at") or row.get("updated_at"))
|
|
for row in rows
|
|
if iso_timestamp(row.get("collected_at") or row.get("updated_at"))
|
|
]
|
|
if not timestamps:
|
|
return {"started_at": "", "ended_at": ""}
|
|
return {"started_at": min(timestamps), "ended_at": max(timestamps)}
|
|
|
|
|
|
def _mean(values: Sequence[float]) -> float:
|
|
if not values:
|
|
return 0.0
|
|
return round(sum(values) / len(values), 3)
|
|
|
|
|
|
def _format_number(value: float) -> str:
|
|
if value.is_integer():
|
|
return str(int(value))
|
|
return f"{value:.3f}".rstrip("0").rstrip(".")
|