"""Read the filled calibration workbook and report what it actually shows. Usage: python calibrate.py [workbook.xlsx] This does three things and refuses to do a fourth. 1. Grades each unit twice — once from Grader A's column, once from Grader B's — and reconciles the pair exactly as a real station would. 2. Reports, per field, how often the graders differed, by how much, and whether that difference changed the derived grade. 3. Reports the observed distribution of each counted field, which is what tells you whether a threshold sits at a natural break or in the middle of the data. It does NOT propose new threshold values. Deciding a threshold is a policy call with a rubric_hash change attached, and it needs a human who has read the distribution and the sample size together. What this script does is make that call informed instead of arbitrary. """ from __future__ import annotations import collections import os import sys from openpyxl import load_workbook from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.utils import get_column_letter HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.join(HERE, "refimpl")) import grading as gr # noqa: E402 DEFAULT_WB = os.path.join(HERE, "sovereign-circuit-batch-1-calibration.xlsx") TRUE = {"TRUE", "T", "YES", "Y", "1", "true", True, 1} FALSE = {"FALSE", "F", "NO", "N", "0", "false", False, 0} class SheetError(Exception): """A filled sheet could not be read. Never guessed around.""" def coerce(fname, raw): """Turn a spreadsheet cell into a typed observation, or refuse.""" cls, kind, tol, desc = gr.FIELDS[fname] if raw is None or (isinstance(raw, str) and not raw.strip()): raise SheetError(f"{fname}: blank. A blank is not an observation — " f"enter the default ({gr.DEFAULTS[fname]!r}) " f"explicitly if it applies.") if kind == "bool": if raw in TRUE: return True if raw in FALSE: return False raise SheetError(f"{fname}: {raw!r} is not TRUE or FALSE") if kind == "int": try: v = int(float(raw)) except (TypeError, ValueError): raise SheetError(f"{fname}: {raw!r} is not a number") if v < 0: raise SheetError(f"{fname}: {v} is negative") return v if kind == "list": s = str(raw).strip() if s.lower() in ("none", "-", "\u2014"): return [] return sorted(x.strip() for x in s.split(",") if x.strip()) if kind == "enum": s = str(raw).strip() if s not in gr._ENUMS[fname]: raise SheetError(f"{fname}: {s!r} not one of " f"{sorted(gr._ENUMS[fname])}") return s s = str(raw).strip() return s def read_observations(ws, n_units): """Return {unit_id: sheet} for every column that has any data in it.""" # Locate the field column and the unit columns from the header row. header = {ws.cell(row=4, column=c).value: c for c in range(1, 6 + n_units)} unit_cols = {k: v for k, v in header.items() if isinstance(k, str) and k.startswith("UNIT-")} if not unit_cols: raise SheetError(f"{ws.title}: no UNIT- columns found in row 4") # Map field name -> row. field_rows = {} for row in range(5, ws.max_row + 1): v = ws.cell(row=row, column=1).value if isinstance(v, str) and v in gr.FIELDS: field_rows[v] = row missing = set(gr.FIELDS) - set(field_rows) if missing: raise SheetError(f"{ws.title}: missing rows for {sorted(missing)}") out = {} for unit, col in sorted(unit_cols.items()): raws = {f: ws.cell(row=r, column=col).value for f, r in field_rows.items()} if all(v is None or (isinstance(v, str) and not v.strip()) for v in raws.values()): continue # column untouched — unit not graded yet sheet = {} errors = [] for f, raw in raws.items(): try: sheet[f] = coerce(f, raw) except SheetError as e: errors.append(str(e)) if errors: out[unit] = ("ERROR", errors) else: out[unit] = ("OK", sheet) return out def main(path=DEFAULT_WB): wb = load_workbook(path, data_only=True) n_units = 20 A = read_observations(wb["Observe A"], n_units) B = read_observations(wb["Observe B"], n_units) units = sorted(set(A) & set(B)) only_a = sorted(set(A) - set(B)) only_b = sorted(set(B) - set(A)) print(f"rubric = {gr.RUBRIC_VERSION}") print(f"rubric_hash = {gr.rubric_hash()}") print(f"workbook = {os.path.basename(path)}") print() if not units: print("No unit has been graded by both graders yet. Fill matching " "columns on 'Observe A' and 'Observe B' and run this again.") if only_a or only_b: print(f" graded by A only: {only_a}") print(f" graded by B only: {only_b}") return 0 if only_a or only_b: print("SINGLE-GRADED UNITS — excluded, a lone reading calibrates nothing:") print(f" A only: {only_a or '—'}") print(f" B only: {only_b or '—'}") print() rows = [] bad = [] field_diffs = collections.Counter() field_deltas = collections.defaultdict(list) pivotal = collections.Counter() dist = collections.defaultdict(collections.Counter) statuses = collections.Counter() tiers = collections.Counter() for unit in units: sa, sb = A[unit], B[unit] if sa[0] == "ERROR" or sb[0] == "ERROR": errs = ([f"A: {e}" for e in (sa[1] if sa[0] == "ERROR" else [])] + [f"B: {e}" for e in (sb[1] if sb[0] == "ERROR" else [])]) bad.append((unit, errs)) continue ha, ib = sa[1], sb[1] rec = gr.reconcile(ha, ib) ga, gb = gr.grade(ha), gr.grade(ib) statuses[rec.status] += 1 differing = [f for f in gr.FIELDS if ha[f] != ib[f]] for f in differing: field_diffs[f] += 1 if gr.FIELDS[f][1] == "int": field_deltas[f].append(abs(int(ha[f]) - int(ib[f]))) piv = [] if rec.status == "void": # A void is caused by the Class M fields that differ, full stop. # minimal_divergence_set answers a grade question and can come back # empty here, because a void does not require the grades to differ. piv = sorted(f for f in differing if gr.FIELDS[f][0] == "M") elif rec.status == "material": try: piv = sorted(gr.minimal_divergence_set(ha, ib)) except Exception: piv = list(differing) for f in piv: pivotal[f] += 1 for f, (cls, kind, tol, desc) in gr.FIELDS.items(): if kind in ("int", "enum"): dist[f][ha[f]] += 1 dist[f][ib[f]] += 1 if rec.status == "void": agreed_tier = "—" # a voided grade has no usable tier else: agreed_tier = ga.tier if ga.code == gb.code else "?" tiers[agreed_tier] += 1 rows.append(dict(unit=unit, a=ga.code, b=gb.code, status=rec.status, tier=agreed_tier, pivotal=piv, ndiff=len(differing))) # ---------------- console report ---------------- print(f"UNITS GRADED BY BOTH: {len(rows)} unreadable: {len(bad)}") print() for unit, errs in bad: print(f" {unit} UNREADABLE:") for e in errs: print(f" {e}") if bad: print() print("PER UNIT") print(f" {'unit':<10} {'A':<10} {'B':<10} {'status':<11} tier ndiff pivotal") for r in rows: print(f" {r['unit']:<10} {r['a']:<10} {r['b']:<10} " f"{r['status']:<11} {r['tier']:<5} {r['ndiff']:<6} " f"{','.join(r['pivotal']) or '—'}") print() print("RECONCILIATION OUTCOMES") n = len(rows) for st in ("agreed", "immaterial", "material", "void"): c = statuses.get(st, 0) print(f" {st:<12} {c:>3} {c / n * 100:5.1f}%") print() if statuses.get("void"): print(" VOIDS are the finding that matters most. A void means the two") print(" graders read one instrument two different ways. That is a") print(" measurement-protocol defect, not a threshold problem, and no") print(" threshold change will fix it. Look at the pivotal Class M") print(" fields below and specify how the reading is taken.") print() print("FIELD DISAGREEMENT — how often the two graders differed") print(f" {'field':<32} {'cls':<4} {'n':<4} {'rate':<7} {'tol':<5} " f"{'max delta':<10} within tol?") any_diff = False for f, c in field_diffs.most_common(): cls, kind, tol, desc = gr.FIELDS[f] deltas = field_deltas.get(f, []) mx = max(deltas) if deltas else "—" if cls == "M": verdict = "N/A — VOIDS" elif kind == "int" and deltas: verdict = "yes" if max(deltas) <= tol else "NO — exceeds tolerance" else: verdict = "n/a" print(f" {f:<32} {cls:<4} {c:<4} {c / n * 100:5.1f}% " f"{tol if cls != 'M' else '—':<5} {str(mx):<10} {verdict}") any_diff = True if not any_diff: print(" none — the two graders agreed on every field of every unit.") print(" On a real batch that is suspicious rather than reassuring:") print(" check that the graders were genuinely blind to each other.") print() print("PIVOTAL FIELDS — differences that actually changed the outcome") if pivotal: for f, c in pivotal.most_common(): cls = gr.FIELDS[f][0] noise = field_diffs[f] print(f" {f:<32} cls {cls} pivotal {c:>3}/{noise:<3} times it differed") print() print(" These are the fields worth spending measurement effort on. A") print(" field that differs constantly but is never pivotal is cheap") print(" noise; a field that is pivotal nearly every time it differs is") print(" carrying the whole grade on one person's eyesight.") else: print(" none — no difference changed a derived grade in this batch.") print() print("OBSERVED DISTRIBUTIONS — counted fields with any spread") shown = 0 for f in gr.FIELDS: if gr.FIELDS[f][1] != "int": continue d = dist[f] if len(d) <= 1: continue vals = sorted(d) line = " ".join(f"{v}:{d[v]}" for v in vals) print(f" {f:<32} {line}") shown += 1 if not shown: print(" every counted field held a single value across the batch —") print(" too little spread to locate a threshold. Widen the batch.") print() print("SAMPLE SIZE") print(f" {n} units graded twice = {n * 2} independent readings.") if n < 20: print(" Below 20 this is a dry run of the process, not calibration.") print(" Use it to fix the observation method. Do not move a threshold.") elif n < 40: print(" 20-39 units can expose measurement-protocol defects and") print(" obviously misplaced band edges. It cannot set a threshold") print(" precisely. Move a threshold only where the distribution shows") print(" an unmistakable break, and record that it was a 20-unit call.") else: print(" Enough to move cosmetic band edges with a recorded rationale.") print(" Storage-failure and hinge-progression thresholds still need") print(" ~100 units and longitudinal data — leave those alone.") print() print(" Any threshold change alters rubric_hash. Rounds graded under the") print(" old hash are not comparable to rounds graded under the new one.") print(" Bump the rubric version, republish vectors, and do not") print(" recalibrate mid-batch.") # ---------------- write Results sheet ---------------- ws = wb["Results"] for row in ws.iter_rows(min_row=5, max_row=max(5, ws.max_row), max_col=8): for cell in row: cell.value = None body = Font(name="Calibri", size=10) mono = Font(name="Consolas", size=9) fills = {"agreed": PatternFill("solid", fgColor="E4F0EA"), "immaterial": PatternFill("solid", fgColor="F2F3F0"), "material": PatternFill("solid", fgColor="FAF1DE"), "void": PatternFill("solid", fgColor="F7E9E9")} NOTES = { "agreed": "Identical sheets. Price window may open.", "immaterial": "Sheets differ but derive the same grade. The conservative " "merge is authoritative; price window may open.", "material": "Sheets derive different grades. Re-read the pivotal fields " "before any price round.", "void": "Class M conflict — one instrument, two readings. Re-read the " "unit. No price round may open on this grade.", } for i, r in enumerate(rows): row = 5 + i vals = [r["unit"], r["a"], r["b"], r["status"], r["tier"], ", ".join(r["pivotal"]) or "—", r["ndiff"], NOTES[r["status"]]] for c, v in enumerate(vals, start=1): cell = ws.cell(row=row, column=c, value=v) cell.font = mono if c in (1, 2, 3, 6) else body cell.fill = fills[r["status"]] cell.alignment = Alignment(vertical="top", wrap_text=(c == 8)) for unit, errs in bad: row = 5 + len(rows) + bad.index((unit, errs)) ws.cell(row=row, column=1, value=unit).font = mono ws.cell(row=row, column=4, value="unreadable").font = body ws.cell(row=row, column=8, value="; ".join(errs)).font = body ws.cell(row=row, column=8).alignment = Alignment(wrap_text=True, vertical="top") wb.save(path) print() print(f"Results sheet updated in {os.path.basename(path)}") return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1] if len(sys.argv) > 1 else DEFAULT_WB))