eval.py

#!/usr/bin/env python3
"""Score clock-time predictions against hand labels. Error is in MINUTES.

    ./eval.py --labels clockface-real/labels.jsonl --pred preds.jsonl

Labels (JSONL, one object per line, written by tools/label_server.py):
    {"id": "cf_0001", "time": "3:47", "unsure": false, "unreadable": false}

Predictions (JSONL):
    {"id": "cf_0001", "time": "3:45"}
    {"id": "cf_0001", "minutes": 225.0, "agreement_minutes": 1.2}

Rules this script enforces, because they are the ones that quietly go wrong:
  * Every label must have a prediction. A missing prediction is a failure,
    never a silently dropped row. Use --allow-missing to score anyway; the
    headline then counts each missing item at the worst possible error (360).
  * Items labelled `unreadable` are excluded and the count is printed.
  * Items labelled `unsure` are included by default and also reported alone,
    so you can see whether the label noise is carrying the result.
  * Synthetic or fixture items (source != "real") are excluded from the
    headline unless --allow-synthetic. Numbers in the model card come from
    real photos.
"""

from __future__ import annotations

import argparse
import json
import math
import sys

import clocktime as ct
import version as ver


# ---------------------------------------------------------------- loading

def read_jsonl(path):
    rows = []
    with open(path) as fh:
        for lineno, line in enumerate(fh, 1):
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            try:
                rows.append(json.loads(line))
            except json.JSONDecodeError as exc:
                raise SystemExit(f"{path}:{lineno}: bad JSON: {exc}")
    return rows


def record_minutes(rec, path, what):
    """Pull a face position out of a label or prediction record."""
    if "minutes" in rec and rec["minutes"] is not None:
        return ct.to_minutes(0, float(rec["minutes"]))
    if "time" in rec and rec["time"]:
        return ct.parse(str(rec["time"]))
    if "hour" in rec and "minute" in rec:
        return ct.to_minutes(float(rec["hour"]), float(rec["minute"]))
    raise SystemExit(f"{path}: {what} {rec.get('id')!r} has no time/minutes/hour+minute")


def load_labels(path, allow_synthetic=False, allow_sources=None):
    labels, unreadable, nonreal, seen = {}, [], [], set()
    for rec in read_jsonl(path):
        rid = rec.get("id")
        if not rid:
            raise SystemExit(f"{path}: label with no id: {rec}")
        if rid in seen:
            raise SystemExit(f"{path}: duplicate label id {rid!r}")
        seen.add(rid)
        if rec.get("unreadable"):
            if rec.get("time"):
                print(f"warning: {rid} is marked unreadable but carries the time "
                      f"{rec['time']!r}; excluding it. Re-label it.", file=sys.stderr)
            unreadable.append(rid)
            continue
        src = rec.get("source", "real")
        ok = (src == "real") or allow_synthetic or any(
            src.startswith(p) for p in (allow_sources or []))
        if not ok:
            nonreal.append(rid)
            continue
        labels[rid] = {
            "minutes": record_minutes(rec, path, "label"),
            "unsure": bool(rec.get("unsure")),
            "source": rec.get("source", "real"),
        }
    return labels, unreadable, nonreal


def load_preds(path):
    preds, seen = {}, set()
    for rec in read_jsonl(path):
        rid = rec.get("id")
        if not rid:
            raise SystemExit(f"{path}: prediction with no id: {rec}")
        if rid in seen:
            raise SystemExit(f"{path}: duplicate prediction id {rid!r}")
        seen.add(rid)
        preds[rid] = {
            "minutes": record_minutes(rec, path, "prediction"),
            "agreement_minutes": rec.get("agreement_minutes"),
        }
    return preds


# ---------------------------------------------------------------- scoring

def percentile(sorted_vals, q):
    if not sorted_vals:
        return float("nan")
    if len(sorted_vals) == 1:
        return sorted_vals[0]
    pos = q / 100.0 * (len(sorted_vals) - 1)
    lo = math.floor(pos)
    hi = math.ceil(pos)
    return sorted_vals[lo] + (sorted_vals[hi] - sorted_vals[lo]) * (pos - lo)


def summarise(errors):
    """Everything is in minutes. No degrees, no normalised anything."""
    n = len(errors)
    if n == 0:
        return {"n": 0}
    s = sorted(errors)
    within = lambda t: sum(1 for e in errors if e <= t) / n
    return {
        "n": n,
        "mae_minutes": sum(errors) / n,
        "median_minutes": percentile(s, 50),
        "p90_minutes": percentile(s, 90),
        "max_minutes": s[-1],
        "within_1min": within(1.0),
        "within_3min": within(3.0),
        "within_5min": within(5.0),
        "within_10min": within(10.0),
        "gross_fail_rate": sum(1 for e in errors if e > 30.0) / n,
    }


def evaluate(labels, preds, missing_error=ct.MAX_ERR, allow_missing=False):
    items, missing = [], []
    for rid, lab in labels.items():
        p = preds.get(rid)
        if p is None:
            missing.append(rid)
            if allow_missing:
                items.append({
                    "id": rid, "label": lab["minutes"], "pred": None,
                    "error": missing_error, "unsure": lab["unsure"],
                    "agreement": None, "missing": True,
                })
            continue
        err = ct.error_minutes(p["minutes"], lab["minutes"])
        items.append({
            "id": rid, "label": lab["minutes"], "pred": p["minutes"],
            "error": err, "unsure": lab["unsure"],
            "agreement": p["agreement_minutes"], "missing": False,
            "error_if_hands_swapped": ct.error_minutes(ct.swapped(p["minutes"]), lab["minutes"]),
        })
    extra = sorted(set(preds) - set(labels))
    return items, missing, extra


def risk_coverage(items):
    """MAE when you keep only the most confident fraction of predictions.

    Confidence is the model's hour/minute-hand disagreement in minutes: small
    disagreement means the two hands tell the same story. A useful signal makes
    this table fall as coverage drops.
    """
    scored = [i for i in items if i.get("agreement") is not None and not i["missing"]]
    if len(scored) < 4:
        return None
    scored.sort(key=lambda i: i["agreement"])
    out = []
    for cov in (1.0, 0.9, 0.75, 0.5, 0.25):
        k = max(1, int(round(cov * len(scored))))
        errs = [i["error"] for i in scored[:k]]
        out.append({
            "coverage": k / len(scored),
            "n": k,
            "mae_minutes": sum(errs) / k,
            "within_5min": sum(1 for e in errs if e <= 5.0) / k,
        })
    return out


# ---------------------------------------------------------------- report

def pct(x):
    return f"{100.0 * x:5.1f}%"


def print_block(title, s):
    print(f"\n{title}")
    if s["n"] == 0:
        print("  (no items)")
        return
    print(f"  n                {s['n']}")
    print(f"  MAE              {s['mae_minutes']:7.2f} min")
    print(f"  median           {s['median_minutes']:7.2f} min")
    print(f"  p90              {s['p90_minutes']:7.2f} min")
    print(f"  worst            {s['max_minutes']:7.2f} min")
    print(f"  within  1 min    {pct(s['within_1min'])}")
    print(f"  within  3 min    {pct(s['within_3min'])}")
    print(f"  within  5 min    {pct(s['within_5min'])}")
    print(f"  within 10 min    {pct(s['within_10min'])}")
    print(f"  worse than 30    {pct(s['gross_fail_rate'])}")


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--labels", default="clockface-real/labels.jsonl")
    ap.add_argument("--pred", help="predictions JSONL")
    ap.add_argument("--allow-missing", action="store_true",
                    help="score anyway; missing predictions count as 360 min errors")
    ap.add_argument("--allow-synthetic", action="store_true",
                    help="include items whose label source is not 'real'")
    ap.add_argument("--allow-source", action="append", metavar="PREFIX",
                    help="also score labels whose source starts with PREFIX, e.g. "
                         "--allow-source hf: to score third-party real photos. The "
                         "report says which sources were included.")
    ap.add_argument("--exclude-unsure", action="store_true")
    ap.add_argument("--per-item", action="store_true", help="print every item, worst first")
    ap.add_argument("--json", dest="json_out", help="write the full report here")
    ap.add_argument("--synth", type=int,
                    help="how many synthetic images the model was trained on; "
                         "recorded in the provenance stamp")
    ap.add_argument("--version", dest="version_str",
                    help="release version YYWWNN; defaults to the next one for this week")
    ap.add_argument("--self-test", action="store_true", help="check the metric itself")
    args = ap.parse_args(argv)

    if args.self_test:
        return self_test()
    if not args.pred:
        ap.error("--pred is required (or use --self-test)")

    labels, unreadable, nonreal = load_labels(args.labels, args.allow_synthetic, args.allow_source)
    preds = load_preds(args.pred)
    if not labels:
        why = f"{len(unreadable)} unreadable, {len(nonreal)} not marked source='real'"
        raise SystemExit(
            f"{args.labels}: no usable labels ({why}).\n"
            f"Numbers in the model card come from real photos. Pass --allow-synthetic "
            f"only when you are deliberately scoring something else.")

    items, missing, extra = evaluate(labels, preds, allow_missing=args.allow_missing)
    if missing and not args.allow_missing:
        head = ", ".join(missing[:10]) + (" ..." if len(missing) > 10 else "")
        raise SystemExit(
            f"{len(missing)} of {len(labels)} labelled items have no prediction: {head}\n"
            f"Fix the predictor, or pass --allow-missing to score them as failures.")

    scored = items if not args.exclude_unsure else [i for i in items if not i["unsure"]]
    errors = [i["error"] for i in scored]

    provenance = ver.stamp(args.version_str, args.synth, len(scored))
    print(provenance)
    if "-dirty" in provenance:
        print("           working tree is dirty: this number cannot be reproduced from a commit")
    srcs = sorted({v["source"] for v in labels.values()})
    print(f"labels     {args.labels}")
    if srcs != ["real"]:
        print(f"SOURCES    {', '.join(srcs)}")
        print("           not the real test set: these are third-party labels, "
              "reported separately and never as the headline number")
    print(f"predictions {args.pred}")
    print(f"scored     {len(scored)} items"
          f"   (excluded: {len(unreadable)} unreadable, {len(nonreal)} non-real"
          f"{', ' + str(len(items) - len(scored)) + ' unsure' if args.exclude_unsure else ''})")
    if missing:
        print(f"MISSING    {len(missing)} predictions counted at {ct.MAX_ERR:.0f} min each")
    if extra:
        print(f"note       {len(extra)} predictions have no label; ignored")

    print_block("ALL SCORED ITEMS  (this is the number that goes in the model card)",
                summarise(errors))

    confident = [i["error"] for i in scored if not i["unsure"]]
    unsure = [i["error"] for i in scored if i["unsure"]]
    if unsure and not args.exclude_unsure:
        print_block(f"labels marked confident ({len(confident)})", summarise(confident))
        print_block(f"labels marked unsure ({len(unsure)})", summarise(unsure))

    rc = risk_coverage(items)
    if rc:
        print("\nHAND-AGREEMENT CONFIDENCE  (keep only the most confident predictions)")
        print("  coverage      n     MAE min   within 5 min")
        for r in rc:
            print(f"  {pct(r['coverage'])}  {r['n']:5d}   {r['mae_minutes']:8.2f}   {pct(r['within_5min'])}")

    bad = [i for i in scored if i["error"] > 30.0 and not i["missing"]]
    swap_fixes = [i for i in bad if i.get("error_if_hands_swapped", 999) < i["error"] - 15]
    if bad:
        print(f"\nDIAGNOSTIC  {len(bad)} items worse than 30 min; "
              f"{len(swap_fixes)} of those would improve by swapping the hands")

    if args.per_item:
        print("\nPER ITEM (worst first)")
        for i in sorted(scored, key=lambda i: -i["error"]):
            p = "MISSING" if i["missing"] else ct.fmt(i["pred"])
            flag = " unsure" if i["unsure"] else ""
            print(f"  {i['id']:<12} label {ct.fmt(i['label']):>6}  pred {p:>7}"
                  f"  err {i['error']:7.2f} min{flag}")

    if args.json_out:
        report = {
            "provenance": provenance,
            "version": args.version_str or ver.next_version(),
            "code": ver.code_hash(),
            "synth_images": args.synth,
            "labels_path": args.labels, "pred_path": args.pred,
            "n_labels": len(labels), "n_scored": len(scored),
            "n_unreadable_excluded": len(unreadable), "n_nonreal_excluded": len(nonreal),
            "n_missing_predictions": len(missing),
            "headline": summarise(errors),
            "confident_only": summarise(confident),
            "unsure_only": summarise(unsure),
            "risk_coverage": rc,
            "items": scored,
        }
        with open(args.json_out, "w") as fh:
            json.dump(report, fh, indent=2)
        print(f"\nwrote {args.json_out}")
    return 0


# ---------------------------------------------------------------- self-test

def self_test():
    """Assertions on the metric. Run this whenever clocktime.py changes."""
    checks = []

    def check(desc, got, want, tol=1e-6):
        ok = abs(got - want) <= tol
        checks.append(ok)
        print(f"  {'ok  ' if ok else 'FAIL'} {desc:<52} got {got:8.3f}  want {want:8.3f}")

    e = lambda a, b: ct.error_minutes(ct.parse(a), ct.parse(b))
    print("metric self-test (all values in minutes)")
    check("identical times", e("3:47", "3:47"), 0)
    check("one minute apart", e("3:47", "3:48"), 1)
    check("across the 12 seam 11:58 vs 12:02", e("11:58", "12:02"), 4)
    check("across the 12 seam 12:02 vs 11:58", e("12:02", "11:58"), 4)
    check("opposite sides of the face", e("12:00", "6:00"), 360)
    check("never exceeds 360", e("12:00", "6:01"), 359)
    check("24h clock reads the same face", e("15:47", "3:47"), 0)
    check("midnight is noon on a face", e("00:00", "12:00"), 0)
    check("wrong hour, right minute", e("4:15", "3:15"), 60)
    check("bare digits parse", e("347", "3:47"), 0)
    check("bare digits parse 4-digit", e("1215", "12:15"), 0)
    check("hand swap 3:00 -> 12:15", ct.error_minutes(ct.swapped(ct.parse("3:00")), ct.parse("12:15")), 0)
    # Swapping is not an involution: at 12:15 the hour hand is 15 face-minutes
    # past 12, which read as a minute hand is 15/12 = 1.25 minutes.
    check("hand swap 12:15 -> 3:01.25", ct.swapped(ct.parse("12:15")), ct.to_minutes(3, 1.25))
    # 12:15 -> 3:01.25 -> 12:15.104, so swapping twice does not quite return.
    check("swap of a swap is not the original",
          ct.error_minutes(ct.swapped(ct.swapped(ct.parse("12:15"))), ct.parse("12:15")), 15 / 144)
    check("hand swap is a no-op at 12:00", ct.error_minutes(ct.swapped(ct.parse("12:00")), ct.parse("12:00")), 0)

    print("\nsummary statistics on a known set")
    errs = [0.0, 1.0, 2.0, 3.0, 100.0]
    s = summarise(errs)
    check("MAE of [0,1,2,3,100]", s["mae_minutes"], 21.2)
    check("median of [0,1,2,3,100]", s["median_minutes"], 2.0)
    check("within 3 min of [0,1,2,3,100]", s["within_3min"], 0.8)
    check("gross fail rate", s["gross_fail_rate"], 0.2)
    check("p90", s["p90_minutes"], 61.2)

    print("\nround trip through parse/format")
    for t in ["12:00", "1:05", "6:30", "11:59", "3:47"]:
        got = ct.fmt(ct.parse(t))
        ok = got == t
        checks.append(ok)
        print(f"  {'ok  ' if ok else 'FAIL'} {t} -> {got}")

    bad = 0
    for junk in ["", "abc", "3:60", "25:00", "3:", ":47"]:
        try:
            ct.parse(junk)
            print(f"  FAIL accepted junk {junk!r}")
            checks.append(False)
        except ValueError:
            bad += 1
    checks.append(True)
    print(f"  ok   rejected {bad} malformed inputs")

    print("\nversion scheme")
    import datetime as _dt
    for d, want in [(_dt.date(2026, 9, 6), "2636"), (_dt.date(2024, 12, 30), "2501"),
                    (_dt.date(2027, 1, 1), "2653"), (_dt.date(2021, 1, 1), "2053")]:
        got = ver.week_stamp(d)
        ok = got == want
        checks.append(ok)
        print(f"  {'ok  ' if ok else 'FAIL'} {d} -> {got} (want {want}, ISO year not calendar year)")
    p = ver.parse("263601")
    ok = (p["iso_year"], p["iso_week"], p["release"], p["week_starts"]) == (2026, 36, 1, "2026-08-31")
    checks.append(ok)
    print(f"  {'ok  ' if ok else 'FAIL'} 263601 parses to week 36 of 2026, starting 2026-08-31")
    try:
        ver.parse("26xx01"); checks.append(False); print("  FAIL accepted junk version")
    except ValueError:
        checks.append(True); print("  ok   rejected a malformed version")

    n_fail = sum(1 for c in checks if not c)
    print(f"\n{len(checks) - n_fail}/{len(checks)} checks passed")
    return 1 if n_fail else 0


if __name__ == "__main__":
    sys.exit(main())