swarm / companies/sparrow/seed/judge.py

#!/usr/bin/env python3
"""Sparrow judge — headless, behavioral, the real quality gate.

Runs the pipeline on a fixed sweep of synthetic frames (same recipe as
`scene.py frames`), measures what the product promises, and prints
PASS/FAIL lines that name the owning file (they feed the build-gate mail).

    pixi run judge                # or: pixi run python judge.py — prints + judge_report.md
Exit code 0 only when every check is green.
"""
import json
import sys
from pathlib import Path

import cv2
import numpy as np

ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT))

import core  # noqa: E402
import scene  # noqa: E402

# Product-grade thresholds the SPRINT must hit (seed starts failing on
# purpose — that is the backlog). Raise with care, in a PR, with a talk.
THRESHOLDS = {
    "feature_density": 130,   # avg features/frame — features.py owner
    "tracking_continuity": 0.90,  # fraction of tracks surviving 2+ frames — tracker.py
    "jitter_px": 1.5,        # median frame-to-frame residual, px — tracker.py
    "map_growth": [50, 900],  # sparse map point count band at t=3s — map.py
}


def density_check(frame):
    pts, _ = core.extract(frame)
    return len(pts)


def run():
    lines = []
    nsamp = 40
    density = []
    pipe = core.Pipeline()
    continuity_survive = continuity_total = 0
    res_hist = []
    prev_pts = None
    prev_gray = None
    import core as _c
    for i in range(nsamp):
        t = i / 10.0
        f = scene.render(t)
        res = pipe.on_frame(f)
        density.append(res["features"])
        gray = cv2.cvtColor(f, cv2.COLOR_BGR2GRAY)
        # plain continuity stats via the seed flow for reporting
        if prev_pts is not None and len(prev_pts) >= 6:
            cur, matched, lost = core.step(prev_gray, gray, prev_pts)
            continuity_survive += len(cur)
            continuity_total += len(prev_pts)
        pts, _ = core.extract(f)
        if prev_pts is not None and len(prev_pts) >= 4:
            cur, matched_prev, lost = core.step(prev_gray, gray, prev_pts[-len(prev_pts):])
            d = np.linalg.norm(cur[: len(matched_prev)] - matched_prev, axis=1) if len(cur) else np.array([9e9])
            res_hist.append(float(np.median(d)))
        prev_pts = pts.astype(np.float32) if len(pts) else None
        prev_gray = gray

    avg_density = float(np.mean(density))
    cont = continuity_survive / max(1, continuity_total)
    jitter = float(np.median(res_hist)) if res_hist else 9e9
    snap = pipe.map.snapshot()
    map_cnt = snap.get("count", 0)

    def check(name, metric, thresh, owner):
        if isinstance(thresh, list):
            ok = thresh[0] <= metric <= thresh[1]
            verdict = f"{metric:.0f} outside [{thresh[0]}, {thresh[1]}]"
        else:
            ok = metric >= thresh if name != "jitter_px" else metric <= thresh
            verdict = f"{metric:.2f} vs {'<=' if name == 'jitter_px' else '>='} {thresh}"
        lines.append(
            (f"PASS {name}" if ok else f"FAIL {owner}: {name}: {verdict} ")
        )
        return ok

    check("feature_density", avg_density, THRESHOLDS["feature_density"], "features.py")
    check("tracking_continuity", cont, THRESHOLDS["tracking_continuity"], "tracker.py")
    check("jitter_px", jitter, THRESHOLDS["jitter_px"], "tracker.py")
    check("map_growth", float(map_cnt), THRESHOLDS["map_growth"], "map.py")

    fail = sum(1 for l in lines if l.startswith("FAIL"))
    report = "\n".join(lines)
    (ROOT / "judge_report.md").write_text(f"# judge (seed)\n\n{report}\n")
    print(f"sparrow judge: {len(lines) - fail} pass, {fail} fail")
    for l in lines:
        print(("  " + l))
    return 1 if fail else 0


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