`scratch_fuzz_scan.py`

101 lines
#!/usr/bin/env python3
"""Read-only fuzz-cross scanner for the published pil-mid-heal cohort.
Detects the band-blur 'fuzzy cross' signature (intermediate-color pixels
concentrated in a center cross band). $0, local PIL+numpy. Writes a report.
NOTE: scratch tool — does NOT modify any DB row, PNG, or designs.json.
"""
import os, sys, json, subprocess, time
import numpy as np
from PIL import Image

GEN_DIR = "/Users/macstudio3/Projects/wallco-ai/data/generated"
OUT = "/Users/macstudio3/.claude/yolo-queue/output/13-fuzzy-cohort-scan-2026-05-26.md"
BAND_PX = 14            # half-width of the center cross band
INTER_T = 45.0         # RGB euclidean dist beyond which a px is 'intermediate' (between the 2 tones)
DOWN = 256             # downsample edge for palette detection

def dominant_two(img_small):
    q = img_small.convert("RGB").quantize(colors=2, method=Image.MEDIANCUT)
    pal = q.getpalette()[:6]
    return np.array([[pal[0], pal[1], pal[2]], [pal[3], pal[4], pal[5]]], dtype=np.float32)

def scan(path):
    im = Image.open(path).convert("RGB")
    W, H = im.size
    small = im.copy(); small.thumbnail((DOWN, DOWN))
    pal = dominant_two(small)
    a = np.asarray(im, dtype=np.float32)            # H,W,3
    d0 = np.sqrt(((a - pal[0]) ** 2).sum(axis=2))
    d1 = np.sqrt(((a - pal[1]) ** 2).sum(axis=2))
    inter = (np.minimum(d0, d1) > INTER_T)          # H,W bool
    cx, cy = W // 2, H // 2
    band = np.zeros((H, W), dtype=bool)
    band[:, max(0, cx - BAND_PX):cx + BAND_PX] = True
    band[max(0, cy - BAND_PX):cy + BAND_PX, :] = True
    band_ratio = inter[band].mean() if band.any() else 0.0
    rest_ratio = inter[~band].mean() if (~band).any() else 0.0
    loc = band_ratio / max(rest_ratio, 0.001)
    return float(band_ratio), float(rest_ratio), float(loc)

def main():
    # calibration controls
    print("=== CALIBRATION ===", flush=True)
    ctrl = {
        "FUZZY(45842)": os.path.join(GEN_DIR, "midheal_21427_1779809020355.png"),
        "CLEAN(21427)": os.path.join(GEN_DIR, "1779280093561_1168136419.cw-manor-saddle.png"),
    }
    for name, p in ctrl.items():
        if os.path.exists(p):
            b, r, l = scan(p)
            print(f"  {name}: band={b:.3f} rest={r:.3f} loc={l:.1f}", flush=True)

    rows = subprocess.run(
        ["psql", "dw_unified", "-tAc",
         "SELECT id, COALESCE(local_path,'') FROM all_designs WHERE is_published=true AND generator='pil-mid-heal' ORDER BY id;"],
        capture_output=True, text=True).stdout.strip().splitlines()
    print(f"=== SCANNING {len(rows)} published pil-mid-heal ===", flush=True)

    results, missing = [], 0
    t0 = time.time()
    for i, line in enumerate(rows):
        if "|" not in line:
            continue
        rid, lp = line.split("|", 1)
        path = lp if (lp and os.path.exists(lp)) else os.path.join(GEN_DIR, os.path.basename(lp))
        if not lp or not os.path.exists(path):
            missing += 1
            continue
        try:
            b, r, l = scan(path)
            results.append({"id": int(rid), "band": b, "rest": r, "loc": l, "file": os.path.basename(path)})
        except Exception as e:
            missing += 1
        if (i + 1) % 200 == 0:
            print(f"  ...{i+1}/{len(rows)}  ({time.time()-t0:.0f}s)", flush=True)

    results.sort(key=lambda x: (x["band"], x["loc"]), reverse=True)
    # localization-driven buckets (45842 ref: band=0.295, loc=294.8). loc is the
    # artificial-blur-band tell; band magnitude alone varies with a design's tone count.
    def sev(x): return x["loc"] >= 20 and x["band"] >= 0.12
    def mod(x): return (not sev(x)) and x["loc"] >= 5 and x["band"] >= 0.05
    severe = [x for x in results if sev(x)]
    moderate = [x for x in results if mod(x)]
    mild = [x for x in results if not sev(x) and not mod(x)]

    with open(OUT, "w") as f:
        f.write("# Fuzzy-cross scan — published pil-mid-heal cohort — 2026-05-26\n\n")
        f.write(f"Scanned: {len(results)} PNGs · missing/err: {missing}\n\n")
        f.write(f"- **SEVERE** (band ≥0.30 & loc ≥5, i.e. ~45842-class): **{len(severe)}**\n")
        f.write(f"- **MODERATE** (band 0.15–0.30): **{len(moderate)}**\n")
        f.write(f"- **MILD/clean** (band <0.15): **{len(mild)}**\n\n")
        f.write("## Top 40 worst offenders\n\n| id | band | rest | loc | file |\n|----|------|------|-----|------|\n")
        for x in results[:40]:
            f.write(f"| {x['id']} | {x['band']:.3f} | {x['rest']:.3f} | {x['loc']:.1f} | {x['file']} |\n")
    json.dump(results, open(OUT.replace(".md", ".json"), "w"))
    print(f"=== DONE: severe={len(severe)} moderate={len(moderate)} mild={len(mild)} missing={missing} ===", flush=True)
    print("report:", OUT, flush=True)
    print("TOP10:", [(x["id"], round(x["band"], 2), round(x["loc"], 1)) for x in results[:10]], flush=True)

if __name__ == "__main__":
    main()