← back to Wallco Ai

scripts/cactus-sample-eval.py

75 lines

#!/usr/bin/env python3
"""cactus-sample-eval.py — measure the quality-redo sample.

Finds the freshly-generated comfy cactus tiles (generator='comfy', created in
the last N minutes), runs the 6-lens edges scan + a local-Ollama vision quality
score on each, and prints a table + the PASS rate against the new criteria
(edge & center ΔE ≤ 5, vision ≥ 70). Read-only — does NOT publish anything.

  python3 scripts/cactus-sample-eval.py [--minutes 60]
"""
import importlib.util, json, os, subprocess, sys, base64, urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
MIN = 60
if '--minutes' in sys.argv: MIN = int(sys.argv[sys.argv.index('--minutes')+1])
VHOST = os.environ.get('OLLAMA_VISION_HOST', '127.0.0.1:11434')
VMODEL = os.environ.get('OLLAMA_VISION_MODEL', 'llava:latest')

_spec = importlib.util.spec_from_file_location('edges_scan', str(ROOT/'scripts'/'edges-scan.py'))
_es = importlib.util.module_from_spec(_spec); _spec.loader.exec_module(_es)

def psql(sql):
    return subprocess.run(['psql','dw_unified','-At','-F','|','-c',sql],capture_output=True,text=True).stdout.strip()

rows = psql(f"""SELECT id, local_path FROM all_designs
  WHERE category ILIKE 'cactus%' AND kind='seamless_tile' AND generator='comfy'
    AND created_at > now() - interval '{MIN} minutes'
  ORDER BY id;""")
rows = [r.split('|',1) for r in rows.split('\n') if r.strip()]
if not rows:
    print(f"No comfy cactus generated in the last {MIN} min."); sys.exit(0)

PROMPT=("Score 0-100 how much this looks like a REAL design from our top catalog lines (Thibaut, Schumacher, "
        "Cole & Son, Scalamandré, Designers Guild, Maya Romanoff, Arte, Brunschwig & Fils, William Morris, "
        "Zuber-grade scenic, DW Bespoke): clean even all-over repeat, cohesive aged/archival palette, no AI "
        "smears/garbled motifs, no harsh seams, no obvious 2x2 grid repeat, refined. 100=premium catalog "
        "product, 0=broken AI junk. Reply ONLY the integer.")
def vision(path):
    try:
        jpg='/tmp/_cse.jpg'
        subprocess.run(['sips','-Z','512','-s','format','jpeg',path,'--out',jpg],
                       capture_output=True)
        b64=base64.b64encode(open(jpg,'rb').read()).decode()
        body=json.dumps({"model":VMODEL,"prompt":PROMPT,"images":[b64],"stream":False,
                         "options":{"temperature":0.1,"num_predict":10}}).encode()
        r=urllib.request.urlopen(urllib.request.Request(f"http://{VHOST}/api/generate",data=body,
            headers={"Content-Type":"application/json"}),timeout=120)
        m=__import__('re').search(r'\d{1,3}', json.load(r).get('response',''))
        return max(0,min(100,int(m.group()))) if m else None
    except Exception as e: return None

print(f"\nEvaluating {len(rows)} sample tiles (edges scan + vision @ {VMODEL})\n")
print(f"{'id':>7} {'edge':>6} {'mid':>6} {'hmid':>6} {'vmid':>6} {'verdict':>5} {'vision':>6}  result")
seam_pass=vis_pass=both=0
for rid,lp in rows:
    if not lp or not os.path.isfile(lp):
        print(f"{rid:>7}  (no PNG on disk)"); continue
    r=_es.scan_path(Path(lp))
    if not r.get('ok'): print(f"{rid:>7}  scan err: {r.get('error')}"); continue
    e=r['edges_max']; m=r['mids_max']; hm=r['lenses']['h_mid']['score']; vm=r['lenses']['v_mid']['score']
    sp = (e<=5 and m<=5)
    v=vision(lp); vp = (v is not None and v>=70)
    if sp: seam_pass+=1
    if vp: vis_pass+=1
    if sp and vp: both+=1
    flag = 'KEEP' if (sp and vp) else ('seam-fail' if not sp else 'vision-low')
    print(f"{rid:>7} {e:>6.1f} {m:>6.1f} {hm:>6.1f} {vm:>6.1f} {r['verdict']:>5} {str(v):>6}  {flag}")
n=len(rows)
print(f"\n=== SAMPLE RESULT ({n} tiles) ===")
print(f"  seamless PASS (edge & center ΔE≤5): {seam_pass}/{n} ({100*seam_pass//max(n,1)}%)")
print(f"  vision ≥70                        : {vis_pass}/{n}")
print(f"  KEEP (pass BOTH criteria)         : {both}/{n}")
print(f"  vs OLD cactus seamless PASS rate  : ~19% (replicate, no circular padding)")