Spaces:
Running
Running
File size: 6,134 Bytes
54f0281 469c3b7 54f0281 469c3b7 54f0281 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """SPEC §10.1 learner metrics from parsed attempt records (services.parse_attempt). Every block reports n."""
from __future__ import annotations
from collections import Counter
from collections.abc import Sequence
from typing import Any
from backend.app import config
from backend.app.analytics.blindspot_map import FOUND, blindspot_points
from backend.app.analytics.calibration import calibration
from backend.app.analytics.froc import froc_curve
from backend.app.search.misstype import BUCKET
BUCKETS = ("search", "recognition", "decision", "interpretation", "overcall")
LOCALIZED = ("found", "mislabeled")
def _focal_results(r: dict[str, Any]) -> list[str]:
res = {o["target"]: o["result"] for o in r["outcomes"]}
return [res[f["id"]] for f in r["findings"] if f["kind"] == "focal" and f["id"] in res]
def miss_counts(r: dict[str, Any]) -> Counter:
return Counter(BUCKET[o["result"]] for o in r["outcomes"] if o["result"] in BUCKET)
def case_level_stats(records: Sequence[dict[str, Any]]) -> dict[str, Any]:
abn = [r for r in records if not r["is_normal"]]
nor = [r for r in records if r["is_normal"]]
def detected(r):
return any(o["result"] in (*LOCALIZED, "pattern_found") for o in r["outcomes"])
def clean(r):
return not any(o["result"] in ("false_positive", "pattern_false") for o in r["outcomes"])
focal = [x for r in records for x in _focal_results(r)]
fp = sum(o["result"] == "false_positive" for r in records for o in r["outcomes"])
mix = Counter()
for r in records:
mix.update(miss_counts(r))
return {
"n": len(records),
"n_abnormal": len(abn),
"n_normal": len(nor),
"sensitivity": sum(map(detected, abn)) / len(abn) if abn else None,
"specificity": sum(map(clean, nor)) / len(nor) if nor else None,
"localization_fraction": sum(x in LOCALIZED for x in focal) / len(focal) if focal else None,
"n_focal_findings": len(focal),
"false_positives_per_image": fp / len(records) if records else 0.0,
"miss_type_mix": {b: int(mix.get(b, 0)) for b in BUCKETS},
"score_mean": round(sum(r["score"] for r in records) / len(records), 1) if records else 0.0,
}
def learning_curve(records: Sequence[dict[str, Any]], window: int = 10) -> dict[str, Any]:
pts = []
for i in range(len(records)):
w = records[max(0, i - window + 1) : i + 1]
pts.append(
{
"attempt": i + 1,
"success_rate": sum(r["success"] for r in w) / len(w),
"window_n": len(w),
"score": records[i]["score"],
}
)
per_label: dict[str, list[dict[str, Any]]] = {}
seq: dict[str, list[float]] = {}
for r in records:
res = {o["target"]: o["result"] for o in r["outcomes"]}
for f in r["findings"]:
ok = res.get(f["id"]) in (*LOCALIZED, "pattern_found")
seq.setdefault(f["label"], []).append(1.0 if ok else 0.0)
if r["is_normal"]:
seq.setdefault("normal", []).append(1.0 if r["success"] else 0.0)
for lab, s in seq.items():
per_label[lab] = [
{
"attempt": i + 1,
"success_rate": sum(s[max(0, i - window + 1) : i + 1]) / len(s[max(0, i - window + 1) : i + 1]),
}
for i in range(len(s))
]
return {"window": window, "n": len(records), "overall": pts, "per_label": per_label}
def miss_type_windows(records: Sequence[dict[str, Any]], size: int = 10) -> dict[str, Any]:
out = []
for start in range(0, len(records), size):
chunk = records[start : start + size]
c = Counter()
for r in chunk:
c.update(miss_counts(r))
out.append(
{"from": start + 1, "to": start + len(chunk), "n": len(chunk), **{b: int(c.get(b, 0)) for b in BUCKETS}}
)
return {"window": size, "n": len(records), "windows": out}
def review_area_habit(records: Sequence[dict[str, Any]]) -> dict[str, Any]:
areas = config.review_area_ids()
n = len(records)
out = []
for a in areas:
k = sum(a in (r.get("search") or {}).get("visited_review_areas", []) for r in records)
out.append({"area": a, "human": config.zone_human(a), "visited_pct": round(100 * k / n, 1) if n else None})
return {"n": n, "areas": out}
def misses_by_zone(records: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
"""Volumetric blind-spot list: per zone (config volumetric zone order, then any other zone seen) the focal
findings graded there and how many were missed. Rows with n = 0 are omitted."""
counts: dict[str, list[int]] = {}
for r in records:
res = {o["target"]: o["result"] for o in r["outcomes"]}
for f in r["findings"]:
if f["kind"] != "focal" or f["id"] not in res:
continue
z = f.get("zone") or "unlabelled"
c = counts.setdefault(z, [0, 0])
c[0] += 1
c[1] += res[f["id"]] not in FOUND
order = list(config.volumetric_zones_cfg()) + [z for z in counts if z not in config.volumetric_zones_cfg()]
return [
{"zone": z, "human": config.zone_human(z), "n": counts[z][0], "n_missed": counts[z][1]}
for z in order
if z in counts
]
def learner_dashboard(
records: Sequence[dict[str, Any]], abilities: dict[str, tuple[float, int]] | None = None
) -> dict[str, Any]:
recs = sorted(records, key=lambda r: r["submitted_at"] or "")
return {
"n_attempts": len(recs),
"summary": case_level_stats(recs),
"learning_curve": learning_curve(recs),
"miss_type_mix": miss_type_windows(recs),
"calibration": calibration(recs),
"froc": froc_curve(recs),
"blindspot_map": blindspot_points(recs),
"review_area_habit": review_area_habit(recs),
"abilities": [{"label": k, "theta": round(t, 3), "n": n} for k, (t, n) in sorted((abilities or {}).items())],
"empty_message": "Read 5 cases to see your first learning curve." if len(recs) < 5 else None,
}
|