pipeline_miami/verify/reconcile.py

"""Base case (ii): served DB vs an independent re-read of the exact ingested file.

Run:  python -m pipeline_miami.verify.reconcile [--metric KEY] [--all]
Writes docs/qa_reports/miami/<metric_key>__validation.{md,json}.
"""
from __future__ import annotations

import argparse
import json
import sys

import yaml

from pipeline_miami.verify import common
from pipeline_miami.verify.connectors import served_db, source_file


def read_source(c: dict, year: str) -> dict:
    reader = getattr(source_file, c["reader"])
    args = dict(c.get("reader_args", {}))
    if c["reader"] == "abc_book":
        return reader(args["source_id"], args["field"])
    if c["reader"] == "srs":
        return reader(year, args["source_id"])
    return reader(year, **args)


def reconcile_one(c: dict) -> dict:
    metric = c["metric_key"]
    tol = float(c.get("tolerance_abs", 0.0))
    served_allow = int(c.get("served_only_allowance", 0))
    source_allow = c.get("source_only_allowance")
    source_allow = int(source_allow) if source_allow is not None else None
    served_all = served_db.fetch(metric)
    matches = 0
    mismatches: list = []
    suppressed = 0
    served_only: list = []
    source_only: list = []
    n_common = 0
    for year in c["years"]:
        source = {k: v for k, v in read_source(c, year).items() if k[1] == year}
        served = {k: v for k, v in served_all.items() if k[1] == year}
        sk, fk = set(served), set(source)
        served_only += sorted(f"{k[0]} {k[1]}" for k in sk - fk)
        source_only += sorted(f"{k[0]} {k[1]}" for k in fk - sk)
        for k in sk & fk:
            n_common += 1
            s, f = served[k], source[k]
            if s["suppressed"] or f["suppressed"]:
                if s["suppressed"] == f["suppressed"]:
                    suppressed += 1
                else:
                    mismatches.append((k, s["value"], f["value"], "suppression flag differs"))
                continue
            if s["value"] is None and f["value"] is None:
                matches += 1
            elif s["value"] is None or f["value"] is None:
                mismatches.append((k, s["value"], f["value"], None))
            elif abs(s["value"] - f["value"]) <= tol + 1e-9:
                matches += 1
            else:
                mismatches.append((k, s["value"], f["value"], s["value"] - f["value"]))
    return {
        "metric_key": metric, "subgroup": "ALL", "population_cut": "n/a", "years": c["years"], "tolerance_abs": tol,
        "n_served": sum(1 for k in served_all if k[1] in c["years"]), "n_common": n_common, "matches": matches, "suppressed": suppressed,
        "served_only_count": len(served_only), "source_only_count": len(source_only),
        "served_only_examples": served_only[:20], "source_only_examples": source_only[:20],
        "agreement_rate": matches / (matches + len(mismatches)) if (matches + len(mismatches)) else None,
        "served_only_allowance": served_allow, "source_only_allowance": source_allow,
        # Served-only cells (the loader wrote a cell the re-read cannot find) fail unless the component
        # documents an allowance; source-only cells (the file has a school the loader does not serve —
        # schools outside the directory universe) fail only above the component's allowance, unlimited by default.
        "status": "PASS" if (matches and not mismatches and len(served_only) <= served_allow and (source_allow is None or len(source_only) <= source_allow)) else "MISMATCH",
        "mismatches": mismatches,
    }


def render(s: dict) -> str:
    out = [f"# Validation (ii) base case — {s['metric_key']}\n", "Served DB cell-by-cell vs an independent re-read of the exact ingested file(s).\n", "## Summary\n"]
    for k in ["metric_key", "years", "tolerance_abs", "n_served", "n_common", "matches", "suppressed", "served_only_count", "served_only_allowance", "source_only_count", "source_only_allowance", "agreement_rate", "status"]:
        out.append(f"- {k}: {s[k]}")
    if s["served_only_examples"]:
        out += ["\n## Served-only examples\n"] + [f"- {e}" for e in s["served_only_examples"]]
    if s["source_only_examples"]:
        out += ["\n## Source-only examples\n"] + [f"- {e}" for e in s["source_only_examples"]]
    if s["mismatches"]:
        out.append(f"\n## Mismatches ({len(s['mismatches'])})\n")
        out += [f"- {k[0]} {k[1]}: served={a} source={b} Δ={d}" for (k, a, b, d) in s["mismatches"][:50]]
    return "\n".join(out)


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--metric")
    ap.add_argument("--all", action="store_true")
    a = ap.parse_args()
    common.ensure_qa_dir()
    comps = yaml.safe_load(open(common.COMPONENTS_YAML))
    if a.metric:
        comps = [c for c in comps if c["metric_key"] == a.metric]
    elif not a.all:
        sys.exit("pass --metric KEY or --all")
    for c in comps:
        s = reconcile_one(c)
        (common.QA_REPORTS_DIR / f"{c['metric_key']}__validation.md").write_text(render(s))
        js = {k: v for k, v in s.items() if k != "mismatches"}
        js["n_mismatches"] = len(s["mismatches"])
        (common.QA_REPORTS_DIR / f"{c['metric_key']}__validation.json").write_text(json.dumps(js, indent=2))
        print(f"{c['metric_key']:36s} {s['status']:9s} common={s['n_common']:>5d} matches={s['matches']:>5d} mismatches={len(s['mismatches']):>4d} suppressed={s['suppressed']:>4d} served_only={s['served_only_count']:>3d} source_only={s['source_only_count']:>3d}")


if __name__ == "__main__":
    main()